


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


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

a_						:       Any      			=				"""
    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 __snake_case      (   UpperCAmelCase_		:		int      ,       UpperCAmelCase_		:		List[Any]      ,       UpperCAmelCase_		:		Union[str, Any]=8	):
 lowerCamelCase_  				=						height // scale_factor**2
 if height % scale_factor**2 != 0:
  new_height += 1
 lowerCamelCase_  				=						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					(					lowercase    ):





  """simple docstring"""




  def __init__(					self    ,						UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase    ,						):







   """simple docstring"""
   super().__init__()

   self.register_modules(
       unet=UpperCamelCase    ,						scheduler=UpperCamelCase    ,						movq=UpperCamelCase    ,						)
   lowerCamelCase_  				=						2 ** (len(self.movq.config.block_out_channels  ) - 1)






  def 						snake_case	(					self    ,						UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase  ):







   """simple docstring"""
   if latents is None:
    lowerCamelCase_  				=						randn_tensor(UpperCamelCase    ,						generator=UpperCamelCase    ,						device=UpperCamelCase    ,						dtype=UpperCamelCase  )
   else:
    if latents.shape != shape:
     raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {shape}'''  )
    lowerCamelCase_  				=						latents.to(UpperCamelCase  )

   lowerCamelCase_  				=						latents * scheduler.init_noise_sigma
   return latents






  def 						snake_case	(					self    ,						UpperCamelCase=0  ):







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

   lowerCamelCase_  				=						torch.device(f'''cuda:{gpu_id}'''  )

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






  def 						snake_case	(					self    ,						UpperCamelCase=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."  )

   lowerCamelCase_  				=						torch.device(f'''cuda:{gpu_id}'''  )

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

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

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






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







   """simple docstring"""
   if not hasattr(self.unet    ,						"_hf_hook"  ):
    return self.device
   for module in self.unet.modules():
    if (
        hasattr(UpperCamelCase    ,						"_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(UpperCamelCase  )
  def __call__(					self    ,						UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase = 512    ,						UpperCamelCase = 512    ,						UpperCamelCase = 100    ,						UpperCamelCase = 4.0    ,						UpperCamelCase = 1    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = "pil"    ,						UpperCamelCase = True    ,						):







   """simple docstring"""
   lowerCamelCase_  				=						self._execution_device

   lowerCamelCase_  				=						guidance_scale > 1.0

   if isinstance(UpperCamelCase    ,						UpperCamelCase  ):
    lowerCamelCase_  				=						torch.cat(UpperCamelCase    ,						dim=0  )
   lowerCamelCase_  				=						image_embeds.shape[0] * num_images_per_prompt
   if isinstance(UpperCamelCase    ,						UpperCamelCase  ):
    lowerCamelCase_  				=						torch.cat(UpperCamelCase    ,						dim=0  )

   if do_classifier_free_guidance:
    lowerCamelCase_  				=						image_embeds.repeat_interleave(UpperCamelCase    ,						dim=0  )
    lowerCamelCase_  				=						negative_image_embeds.repeat_interleave(UpperCamelCase    ,						dim=0  )

   lowerCamelCase_  				=						torch.cat([negative_image_embeds, image_embeds]    ,						dim=0  ).to(dtype=self.unet.dtype    ,						device=UpperCamelCase  )

   self.scheduler.set_timesteps(UpperCamelCase    ,						device=UpperCamelCase  )
   lowerCamelCase_  				=						self.scheduler.timesteps

   lowerCamelCase_  				=						self.unet.config.in_channels

   lowerCamelCase_				,lowerCamelCase_  				=						downscale_height_and_width(UpperCamelCase    ,						UpperCamelCase    ,						self.movq_scale_factor  )

   # create initial latent
   lowerCamelCase_  				=						self.prepare_latents(
       (batch_size, num_channels_latents, height, width)    ,						image_embeds.dtype    ,						UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase    ,						self.scheduler    ,						)

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

    lowerCamelCase_  				=						{"image_embeds": image_embeds}
    lowerCamelCase_  				=						self.unet(
        sample=UpperCamelCase    ,						timestep=UpperCamelCase    ,						encoder_hidden_states=UpperCamelCase    ,						added_cond_kwargs=UpperCamelCase    ,						return_dict=UpperCamelCase    ,						)[0]

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

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

   if output_type == "pil":
    lowerCamelCase_  				=						self.numpy_to_pil(UpperCamelCase  )

   if not return_dict:
    return (image,)

   return ImagePipelineOutput(images=UpperCamelCase  )







'''simple docstring'''

from dataclasses import dataclass
from typing import Optional, Tuple

import torch
from torch import nn
from transformers import RobertaPreTrainedModel, XLMRobertaConfig, XLMRobertaModel
from transformers.utils import ModelOutput





@dataclass
class   snake_case					(					lowercase    ):





  """simple docstring"""




  _lowerCamelCase						=						None
  _lowerCamelCase						=						None
  _lowerCamelCase						=						None
  _lowerCamelCase						=						None





class   snake_case					(					lowercase    ):





  """simple docstring"""




  def __init__(					self    ,						UpperCamelCase=1    ,						UpperCamelCase=0    ,						UpperCamelCase=2    ,						UpperCamelCase=512    ,						UpperCamelCase="cls"    ,						UpperCamelCase=False    ,						UpperCamelCase=True    ,						**UpperCamelCase    ,						):







   """simple docstring"""
   super().__init__(pad_token_id=UpperCamelCase    ,						bos_token_id=UpperCamelCase    ,						eos_token_id=UpperCamelCase    ,						**UpperCamelCase  )
   lowerCamelCase_  				=						project_dim
   lowerCamelCase_  				=						pooler_fn
   lowerCamelCase_  				=						learn_encoder
   lowerCamelCase_  				=						use_attention_mask





class   snake_case					(					lowercase    ):





  """simple docstring"""




  _lowerCamelCase						=						[r"pooler", r"logit_scale"]
  _lowerCamelCase						=						[r"position_ids", r"predictions.decoder.bias"]
  _lowerCamelCase						=						"roberta"
  _lowerCamelCase						=						RobertaSeriesConfig






  def __init__(					self    ,						UpperCamelCase  ):







   """simple docstring"""
   super().__init__(UpperCamelCase  )
   lowerCamelCase_  				=						XLMRobertaModel(UpperCamelCase  )
   lowerCamelCase_  				=						nn.Linear(config.hidden_size    ,						config.project_dim  )
   lowerCamelCase_  				=						getattr(UpperCamelCase    ,						"has_pre_transformation"    ,						UpperCamelCase  )
   if self.has_pre_transformation:
    lowerCamelCase_  				=						nn.Linear(config.hidden_size    ,						config.project_dim  )
    lowerCamelCase_  				=						nn.LayerNorm(config.hidden_size    ,						eps=config.layer_norm_eps  )
   self.post_init()







  def 						snake_case	(					self    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						UpperCamelCase = None    ,						):







   """simple docstring"""
   lowerCamelCase_  				=						return_dict if return_dict is not None else self.config.use_return_dict

   lowerCamelCase_  				=						self.base_model(
       input_ids=UpperCamelCase    ,						attention_mask=UpperCamelCase    ,						token_type_ids=UpperCamelCase    ,						position_ids=UpperCamelCase    ,						head_mask=UpperCamelCase    ,						inputs_embeds=UpperCamelCase    ,						encoder_hidden_states=UpperCamelCase    ,						encoder_attention_mask=UpperCamelCase    ,						output_attentions=UpperCamelCase    ,						output_hidden_states=True if self.has_pre_transformation else output_hidden_states    ,						return_dict=UpperCamelCase    ,						)

   if self.has_pre_transformation:
    lowerCamelCase_  				=						outputs["hidden_states"][-2]
    lowerCamelCase_  				=						self.pre_LN(UpperCamelCase  )
    lowerCamelCase_  				=						self.transformation_pre(UpperCamelCase  )

    return TransformationModelOutput(
        projection_state=UpperCamelCase    ,						last_hidden_state=outputs.last_hidden_state    ,						hidden_states=outputs.hidden_states    ,						attentions=outputs.attentions    ,						)
   else:
    lowerCamelCase_  				=						self.transformation(outputs.last_hidden_state  )
    return TransformationModelOutput(
        projection_state=UpperCamelCase    ,						last_hidden_state=outputs.last_hidden_state    ,						hidden_states=outputs.hidden_states    ,						attentions=outputs.attentions    ,						)



