





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     =    logging.get_logger(__name__)  # pylint: disable=invalid-name

__A     =    "\n    Examples:\n        ```py\n        >>> import torch\n        >>> import numpy as np\n\n        >>> from diffusers import KandinskyV22PriorPipeline, KandinskyV22ControlnetPipeline\n        >>> from transformers import pipeline\n        >>> from diffusers.utils import load_image\n\n\n        >>> def make_hint(image, depth_estimator):\n        ...     image = depth_estimator(image)[\"depth\"]\n        ...     image = np.array(image)\n        ...     image = image[:, :, None]\n        ...     image = np.concatenate([image, image, image], axis=2)\n        ...     detected_map = torch.from_numpy(image).float() / 255.0\n        ...     hint = detected_map.permute(2, 0, 1)\n        ...     return hint\n\n\n        >>> depth_estimator = pipeline(\"depth-estimation\")\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     \"kandinsky-community/kandinsky-2-2-prior\", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior = pipe_prior.to(\"cuda\")\n\n        >>> pipe = KandinskyV22ControlnetPipeline.from_pretrained(\n        ...     \"kandinsky-community/kandinsky-2-2-controlnet-depth\", torch_dtype=torch.float16\n        ... )\n        >>> pipe = pipe.to(\"cuda\")\n\n\n        >>> img = load_image(\n        ...     \"https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main\"\n        ...     \"/kandinsky/cat.png\"\n        ... ).resize((768, 768))\n\n        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to(\"cuda\")\n\n        >>> prompt = \"A robot, 4k photo\"\n        >>> negative_prior_prompt = \"lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature\"\n\n        >>> generator = torch.Generator(device=\"cuda\").manual_seed(43)\n\n        >>> image_emb, zero_image_emb = pipe_prior(\n        ...     prompt=prompt, negative_prompt=negative_prior_prompt, generator=generator\n        ... ).to_tuple()\n\n        >>> images = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     hint=hint,\n        ...     num_inference_steps=50,\n        ...     generator=generator,\n        ...     height=768,\n        ...     width=768,\n        ... ).images\n\n        >>> images[0].save(\"robot_cat.png\")\n        ```\n"





def 					lowerCAmelCase_			(	__a     ,   __a     ,   __a=8		)					->		int:







  """simple docstring"""





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






class        _SCREAMING_SNAKE_CASE						(     __SCREAMING_SNAKE_CASE   ):







 '''simple docstring'''

 def __init__(self    :       List[Any]	,			UpperCAmelCase_    :       UNetaDConditionModel	,			UpperCAmelCase_    :       DDPMScheduler	,			UpperCAmelCase_    :       VQModel	,			)	->int:



   '''simple docstring'''



   super().__init__()

   self.register_modules(
       unet=UpperCAmelCase_	,			scheduler=UpperCAmelCase_	,			movq=UpperCAmelCase_	,			)
   lowerCamelCase__:				List[str]        =2 ** (len(self.movq.config.block_out_channels) - 1)

 def 							SCREAMING_SNAKE_CASE_ (self    :       Tuple	,			UpperCAmelCase_    :       Optional[Any]	,			UpperCAmelCase_    :       Dict	,			UpperCAmelCase_    :       int	,			UpperCAmelCase_    :       Optional[Any]	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       int)	->List[str]:



   '''simple docstring'''



   if latents is None:
     lowerCamelCase__:				Any        =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__:				int        =latents.to(UpperCAmelCase_)

   lowerCamelCase__:				Any        =latents * scheduler.init_noise_sigma
   return latents

 def 							SCREAMING_SNAKE_CASE_ (self    :       str	,			UpperCAmelCase_    :       Any=0)	->int:



   '''simple docstring'''



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

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

   lowerCamelCase__:				List[str]        =[
       self.unet,
       self.movq,
   ]
   for cpu_offloaded_model in models:
     if cpu_offloaded_model is not None:
       cpu_offload(UpperCAmelCase_	,			UpperCAmelCase_)

 def 							SCREAMING_SNAKE_CASE_ (self    :       Dict	,			UpperCAmelCase_    :       Any=0)	->Any:



   '''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__:				List[Any]        =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__:				Optional[int]        =None
   for cpu_offloaded_model in [self.unet, self.movq]:
     lowerCamelCase__:				str        =cpu_offload_with_hook(UpperCAmelCase_	,			UpperCAmelCase_	,			prev_module_hook=UpperCAmelCase_)

   # We'll offload the last model manually.
   lowerCamelCase__:				Optional[int]        =hook

 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def 							SCREAMING_SNAKE_CASE_ (self    :       Dict)	->Union[str, Any]:



   '''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    :       int	,			UpperCAmelCase_    :       Union[torch.FloatTensor, List[torch.FloatTensor]]	,			UpperCAmelCase_    :       Union[torch.FloatTensor, List[torch.FloatTensor]]	,			UpperCAmelCase_    :       torch.FloatTensor	,			UpperCAmelCase_    :       int = 512	,			UpperCAmelCase_    :       int = 512	,			UpperCAmelCase_    :       int = 100	,			UpperCAmelCase_    :       float = 4.0	,			UpperCAmelCase_    :       int = 1	,			UpperCAmelCase_    :       Optional[Union[torch.Generator, List[torch.Generator]]] = None	,			UpperCAmelCase_    :       Optional[torch.FloatTensor] = None	,			UpperCAmelCase_    :       Optional[str] = "pil"	,			UpperCAmelCase_    :       bool = True	,			)	->str:



   '''simple docstring'''



   lowerCamelCase__:				Tuple        =self._execution_device

   lowerCamelCase__:				Optional[Any]        =guidance_scale > 1.0

   if isinstance(UpperCAmelCase_	,			UpperCAmelCase_):
     lowerCamelCase__:				Any        =torch.cat(UpperCAmelCase_	,			dim=0)
   if isinstance(UpperCAmelCase_	,			UpperCAmelCase_):
     lowerCamelCase__:				Dict        =torch.cat(UpperCAmelCase_	,			dim=0)
   if isinstance(UpperCAmelCase_	,			UpperCAmelCase_):
     lowerCamelCase__:				Union[str, Any]        =torch.cat(UpperCAmelCase_	,			dim=0)

   lowerCamelCase__:				Tuple        =image_embeds.shape[0] * num_images_per_prompt

   if do_classifier_free_guidance:
     lowerCamelCase__:				Dict        =image_embeds.repeat_interleave(UpperCAmelCase_	,			dim=0)
     lowerCamelCase__:				int        =negative_image_embeds.repeat_interleave(UpperCAmelCase_	,			dim=0)
     lowerCamelCase__:				Tuple        =hint.repeat_interleave(UpperCAmelCase_	,			dim=0)

   lowerCamelCase__:				Optional[int]        =torch.cat([negative_image_embeds, image_embeds]	,			dim=0).to(dtype=self.unet.dtype	,			device=UpperCAmelCase_)
   lowerCamelCase__:				Optional[int]        =torch.cat([hint, hint]	,			dim=0).to(dtype=self.unet.dtype	,			device=UpperCAmelCase_)
   self.scheduler.set_timesteps(UpperCAmelCase_	,			device=UpperCAmelCase_)
   lowerCamelCase__:				int        =self.scheduler.timesteps

   lowerCamelCase__:				int        =self.movq.config.latent_channels

   lowerCamelCase__:				List[str]        =downscale_height_and_width(UpperCAmelCase_	,			UpperCAmelCase_	,			self.movq_scale_factor)

   # create initial latent
   lowerCamelCase__:				Tuple        =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__:				Optional[int]        =torch.cat([latents] * 2) if do_classifier_free_guidance else latents

     lowerCamelCase__:				List[str]        ={"image_embeds": image_embeds, "hint": hint}
     lowerCamelCase__:				Any        =self.unet(
         sample=UpperCAmelCase_	,			timestep=UpperCAmelCase_	,			encoder_hidden_states=UpperCAmelCase_	,			added_cond_kwargs=UpperCAmelCase_	,			return_dict=UpperCAmelCase_	,			)[0]

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

     # compute the previous noisy sample x_t -> x_t-1
     lowerCamelCase__:				str        =self.scheduler.step(
         UpperCAmelCase_	,			UpperCAmelCase_	,			UpperCAmelCase_	,			generator=UpperCAmelCase_	,			)[0]
   # post-processing
   lowerCamelCase__:				Any        =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__:				Optional[Any]        =image * 0.5 + 0.5
     lowerCamelCase__:				Optional[int]        =image.clamp(0	,			1)
     lowerCamelCase__:				Optional[int]        =image.cpu().permute(0	,			2	,			3	,			1).float().numpy()

   if output_type == "pil":
     lowerCamelCase__:				Optional[Any]        =self.numpy_to_pil(UpperCAmelCase_)

   if not return_dict:
     return (image,)

   return ImagePipelineOutput(images=UpperCAmelCase_)








from argparse import ArgumentParser, Namespace

from ..utils import logging
from . import BaseTransformersCLICommand





def 					lowerCAmelCase_			(	__a		)					->		int:







  """simple docstring"""





  return ConvertCommand(
      args.model_type     ,   args.tf_checkpoint     ,   args.pytorch_dump_output     ,   args.config     ,   args.finetuning_task_name		)


__A     =    "\ntransformers can only be used from the commandline to convert TensorFlow models in PyTorch, In that case, it requires\nTensorFlow to be installed. Please see https://www.tensorflow.org/install/ for installation instructions.\n"






class        _SCREAMING_SNAKE_CASE						(     __SCREAMING_SNAKE_CASE   ):







 '''simple docstring'''

 @staticmethod
 def 							SCREAMING_SNAKE_CASE_ (UpperCAmelCase_    :       ArgumentParser)	->str:



   '''simple docstring'''



   lowerCamelCase__:				Dict        =parser.add_parser(
       "convert"	,			help="CLI tool to run convert model from original author checkpoints to Transformers PyTorch checkpoints."	,			)
   train_parser.add_argument("--model_type"	,			type=UpperCAmelCase_	,			required=UpperCAmelCase_	,			help="Model's type.")
   train_parser.add_argument(
       "--tf_checkpoint"	,			type=UpperCAmelCase_	,			required=UpperCAmelCase_	,			help="TensorFlow checkpoint path or folder.")
   train_parser.add_argument(
       "--pytorch_dump_output"	,			type=UpperCAmelCase_	,			required=UpperCAmelCase_	,			help="Path to the PyTorch saved model output.")
   train_parser.add_argument("--config"	,			type=UpperCAmelCase_	,			default=""	,			help="Configuration file path or folder.")
   train_parser.add_argument(
       "--finetuning_task_name"	,			type=UpperCAmelCase_	,			default=UpperCAmelCase_	,			help="Optional fine-tuning task name if the TF model was a finetuned model."	,			)
   train_parser.set_defaults(func=UpperCAmelCase_)

 def __init__(self    :       List[Any]	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       str	,			*UpperCAmelCase_    :       Optional[int]	,			)	->List[str]:



   '''simple docstring'''



   lowerCamelCase__:				Dict        =logging.get_logger("transformers-cli/converting")

   self._logger.info(F"""Loading model {model_type}""")
   lowerCamelCase__:				Any        =model_type
   lowerCamelCase__:				Optional[int]        =tf_checkpoint
   lowerCamelCase__:				Any        =pytorch_dump_output
   lowerCamelCase__:				Union[str, Any]        =config
   lowerCamelCase__:				str        =finetuning_task_name







 def 							SCREAMING_SNAKE_CASE_ (self    :       Union[str, Any])	->Tuple:



   '''simple docstring'''



   if self._model_type == "albert":
     try:
       from ..models.albert.convert_albert_original_tf_checkpoint_to_pytorch import (
           convert_tf_checkpoint_to_pytorch,
       )
     except ImportError:
       raise ImportError(UpperCAmelCase_)

     convert_tf_checkpoint_to_pytorch(self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output)
   elif self._model_type == "bert":
     try:
       from ..models.bert.convert_bert_original_tf_checkpoint_to_pytorch import (
           convert_tf_checkpoint_to_pytorch,
       )
     except ImportError:
       raise ImportError(UpperCAmelCase_)

     convert_tf_checkpoint_to_pytorch(self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output)
   elif self._model_type == "funnel":
     try:
       from ..models.funnel.convert_funnel_original_tf_checkpoint_to_pytorch import (
           convert_tf_checkpoint_to_pytorch,
       )
     except ImportError:
       raise ImportError(UpperCAmelCase_)

     convert_tf_checkpoint_to_pytorch(self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output)
   elif self._model_type == "t5":
     try:
       from ..models.ta.convert_ta_original_tf_checkpoint_to_pytorch import convert_tf_checkpoint_to_pytorch
     except ImportError:
       raise ImportError(UpperCAmelCase_)

     convert_tf_checkpoint_to_pytorch(self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output)
   elif self._model_type == "gpt":
     from ..models.openai.convert_openai_original_tf_checkpoint_to_pytorch import (
         convert_openai_checkpoint_to_pytorch,
     )

     convert_openai_checkpoint_to_pytorch(self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output)
   elif self._model_type == "transfo_xl":
     try:
       from ..models.transfo_xl.convert_transfo_xl_original_tf_checkpoint_to_pytorch import (
           convert_transfo_xl_checkpoint_to_pytorch,
       )
     except ImportError:
       raise ImportError(UpperCAmelCase_)

     if "ckpt" in self._tf_checkpoint.lower():
       lowerCamelCase__:				Tuple        =self._tf_checkpoint
       lowerCamelCase__:				List[str]        =""
     else:
       lowerCamelCase__:				Any        =self._tf_checkpoint
       lowerCamelCase__:				Dict        =""
     convert_transfo_xl_checkpoint_to_pytorch(
         UpperCAmelCase_	,			self._config	,			self._pytorch_dump_output	,			UpperCAmelCase_)
   elif self._model_type == "gpt2":
     try:
       from ..models.gpta.convert_gpta_original_tf_checkpoint_to_pytorch import (
           convert_gpta_checkpoint_to_pytorch,
       )
     except ImportError:
       raise ImportError(UpperCAmelCase_)

     convert_gpta_checkpoint_to_pytorch(self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output)
   elif self._model_type == "xlnet":
     try:
       from ..models.xlnet.convert_xlnet_original_tf_checkpoint_to_pytorch import (
           convert_xlnet_checkpoint_to_pytorch,
       )
     except ImportError:
       raise ImportError(UpperCAmelCase_)

     convert_xlnet_checkpoint_to_pytorch(
         self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output	,			self._finetuning_task_name)
   elif self._model_type == "xlm":
     from ..models.xlm.convert_xlm_original_pytorch_checkpoint_to_pytorch import (
         convert_xlm_checkpoint_to_pytorch,
     )

     convert_xlm_checkpoint_to_pytorch(self._tf_checkpoint	,			self._pytorch_dump_output)
   elif self._model_type == "lxmert":
     from ..models.lxmert.convert_lxmert_original_tf_checkpoint_to_pytorch import (
         convert_lxmert_checkpoint_to_pytorch,
     )

     convert_lxmert_checkpoint_to_pytorch(self._tf_checkpoint	,			self._pytorch_dump_output)
   elif self._model_type == "rembert":
     from ..models.rembert.convert_rembert_tf_checkpoint_to_pytorch import (
         convert_rembert_tf_checkpoint_to_pytorch,
     )

     convert_rembert_tf_checkpoint_to_pytorch(self._tf_checkpoint	,			self._config	,			self._pytorch_dump_output)
   else:
     raise ValueError(
         "--model_type should be selected in the list [bert, gpt, gpt2, t5, transfo_xl, xlnet, xlm, lxmert]")

