


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


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

__lowerCamelCase												=						'''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")
        >>> pipe_prior.to("cuda")
        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> zero_image_emb = out.negative_image_embeds
        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")
        >>> pipe.to("cuda")
        >>> image = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=50,
        ... ).images
        >>> image[0].save("cat.png")
        ```
'''


def        a__     (					UpperCamelCase_						:      Optional[Any],   UpperCamelCase_						:      Tuple,   UpperCamelCase_						:      Optional[Any]=8						):
   UpperCAmelCase__	:List[str]      =      height // scale_factor**2
   if height % scale_factor**2 != 0:
      new_height += 1
   UpperCAmelCase__	:int      =      width // scale_factor**2
   if width % scale_factor**2 != 0:
      new_width += 1
   return new_height * scale_factor, new_width * scale_factor




class 				UpperCAmelCase			(			_snake_case						):


 def __init__(      self							:	Dict			,       __lowerCamelCase							:	UNetaDConditionModel			,       __lowerCamelCase							:	DDPMScheduler			,       __lowerCamelCase							:	VQModel			,       ):
    super().__init__()

    self.register_modules(
        unet=__lowerCamelCase			,       scheduler=__lowerCamelCase			,       movq=__lowerCamelCase			,       )
    UpperCAmelCase__	:Tuple      =      2 ** (len(self.movq.config.block_out_channels ) - 1)


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any]			,       __lowerCamelCase							:	Optional[int]			,       __lowerCamelCase							:	List[Any]			,       __lowerCamelCase							:	Optional[Any]			,       __lowerCamelCase							:	List[Any]			,       __lowerCamelCase							:	Dict			,       __lowerCamelCase							:	List[Any] ):
    if latents is None:
       UpperCAmelCase__	:Any      =      randn_tensor(__lowerCamelCase			,       generator=__lowerCamelCase			,       device=__lowerCamelCase			,       dtype=__lowerCamelCase )
    else:
       if latents.shape != shape:
          raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {shape}''' )
       UpperCAmelCase__	:str      =      latents.to(__lowerCamelCase )

    UpperCAmelCase__	:str      =      latents * scheduler.init_noise_sigma
    return latents


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any]			,       __lowerCamelCase							:	List[str]=0 ):
    if is_accelerate_available():
       from accelerate import cpu_offload
    else:
       raise ImportError('''Please install accelerate via `pip install accelerate`''' )

    UpperCAmelCase__	:Optional[int]      =      torch.device(f'''cuda:{gpu_id}''' )

    UpperCAmelCase__	:List[Any]      =      [
        self.unet,
        self.movq,
    ]
    for cpu_offloaded_model in models:
       if cpu_offloaded_model is not None:
          cpu_offload(__lowerCamelCase			,       __lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[str]			,       __lowerCamelCase							:	str=0 ):
    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.''' )

    UpperCAmelCase__	:Optional[int]      =      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)

    UpperCAmelCase__	:Optional[int]      =      None
    for cpu_offloaded_model in [self.unet, self.movq]:
       UpperCAmelCase__   ,						UpperCAmelCase__	:int      =      cpu_offload_with_hook(__lowerCamelCase			,       __lowerCamelCase			,       prev_module_hook=__lowerCamelCase )

    # We'll offload the last model manually.
    UpperCAmelCase__	:List[Any]      =      hook


 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[int] ):
    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





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

    UpperCAmelCase__	:Any      =      guidance_scale > 1.0

    if isinstance(__lowerCamelCase			,       __lowerCamelCase ):
       UpperCAmelCase__	:Union[str, Any]      =      torch.cat(__lowerCamelCase			,       dim=0 )
    UpperCAmelCase__	:str      =      image_embeds.shape[0] * num_images_per_prompt
    if isinstance(__lowerCamelCase			,       __lowerCamelCase ):
       UpperCAmelCase__	:Any      =      torch.cat(__lowerCamelCase			,       dim=0 )

    if do_classifier_free_guidance:
       UpperCAmelCase__	:Optional[int]      =      image_embeds.repeat_interleave(__lowerCamelCase			,       dim=0 )
       UpperCAmelCase__	:Optional[Any]      =      negative_image_embeds.repeat_interleave(__lowerCamelCase			,       dim=0 )

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

    self.scheduler.set_timesteps(__lowerCamelCase			,       device=__lowerCamelCase )
    UpperCAmelCase__	:Tuple      =      self.scheduler.timesteps

    UpperCAmelCase__	:int      =      self.unet.config.in_channels

    UpperCAmelCase__   ,						UpperCAmelCase__	:int      =      downscale_height_and_width(__lowerCamelCase			,       __lowerCamelCase			,       self.movq_scale_factor )

    # create initial latent
    UpperCAmelCase__	:List[Any]      =      self.prepare_latents(
        (batch_size, num_channels_latents, height, width)			,       image_embeds.dtype			,       __lowerCamelCase			,       __lowerCamelCase			,       __lowerCamelCase			,       self.scheduler			,       )

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

       UpperCAmelCase__	:List[Any]      =      {'''image_embeds''': image_embeds}
       UpperCAmelCase__	:Optional[Any]      =      self.unet(
           sample=__lowerCamelCase			,       timestep=__lowerCamelCase			,       encoder_hidden_states=__lowerCamelCase			,       added_cond_kwargs=__lowerCamelCase			,       return_dict=__lowerCamelCase			,       )[0]

       if do_classifier_free_guidance:
          UpperCAmelCase__   ,						UpperCAmelCase__	:List[str]      =      noise_pred.split(latents.shape[1]			,       dim=1 )
          UpperCAmelCase__   ,						UpperCAmelCase__	:List[Any]      =      noise_pred.chunk(2 )
          UpperCAmelCase__   ,						UpperCAmelCase__	:List[Any]      =      variance_pred.chunk(2 )
          UpperCAmelCase__	:str      =      noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
          UpperCAmelCase__	: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"]
       ):
          UpperCAmelCase__   ,						UpperCAmelCase__	:Dict      =      noise_pred.split(latents.shape[1]			,       dim=1 )

       # compute the previous noisy sample x_t -> x_t-1
       UpperCAmelCase__	:Union[str, Any]      =      self.scheduler.step(
           __lowerCamelCase			,       __lowerCamelCase			,       __lowerCamelCase			,       generator=__lowerCamelCase			,       )[0]
    # post-processing
    UpperCAmelCase__	:Tuple      =      self.movq.decode(__lowerCamelCase			,       force_not_quantize=__lowerCamelCase )['''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"]:
       UpperCAmelCase__	:Tuple      =      image * 0.5 + 0.5
       UpperCAmelCase__	:Tuple      =      image.clamp(0			,       1 )
       UpperCAmelCase__	:List[str]      =      image.cpu().permute(0			,       2			,       3			,       1 ).float().numpy()

    if output_type == "pil":
       UpperCAmelCase__	:Optional[int]      =      self.numpy_to_pil(__lowerCamelCase )

    if not return_dict:
       return (image,)

    return ImagePipelineOutput(images=__lowerCamelCase )







'''simple docstring'''





import pytest

from datasets.splits import SplitDict, SplitInfo
from datasets.utils.py_utils import asdict


@pytest.mark.parametrize(
    '''split_dict''',   [
        SplitDict(),
        SplitDict({'''train''': SplitInfo(name='''train''',   num_bytes=1_337,   num_examples=42,   dataset_name='''my_dataset'''						)}						),
        SplitDict({'''train''': SplitInfo(name='''train''',   num_bytes=1_337,   num_examples=42						)}						),
        SplitDict({'''train''': SplitInfo()}						),
    ],   )
def        a__     (					UpperCamelCase_						:      SplitDict						):
   UpperCAmelCase__	:List[Any]      =      split_dict._to_yaml_list()
   assert len(UpperCamelCase_						) == len(UpperCamelCase_						)
   UpperCAmelCase__	:Tuple      =      SplitDict._from_yaml_list(UpperCamelCase_						)
   for split_name, split_info in split_dict.items():
      # dataset_name field is deprecated, and is therefore not part of the YAML dump
      UpperCAmelCase__	:Union[str, Any]      =      None
      # the split name of split_dict takes over the name of the split info object
      UpperCAmelCase__	:Union[str, Any]      =      split_name
   assert split_dict == reloaded


@pytest.mark.parametrize(
    '''split_info''',   [SplitInfo(), SplitInfo(dataset_name=UpperCamelCase_						), SplitInfo(dataset_name='''my_dataset'''						)]						)
def        a__     (					UpperCamelCase_						:      List[Any]						):
   # For backward compatibility, we need asdict(split_dict) to return split info dictrionaries with the "dataset_name"
   # field even if it's deprecated. This way old versionso of `datasets` can still reload dataset_infos.json files
   UpperCAmelCase__	:List[Any]      =      asdict(SplitDict({'''train''': split_info}						)						)
   assert "dataset_name" in split_dict_asdict["train"]
   assert split_dict_asdict["train"]["dataset_name"] == split_info.dataset_name



