





"""simple docstring"""
import functools
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

import huggingface_hub.utils as hf_hub_utils
from tqdm import auto as tqdm_lib


_snake_case     :		List[Any]       						= threading.Lock()
_snake_case     :		Optional[logging.Handler]       						= None

_snake_case     :		Optional[Any]       						= {
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
}

_snake_case     :		List[Any]       						= logging.WARNING

_snake_case     :		Optional[int]       						= True






def 					A__				(			):
							A		=  os.getenv("TRANSFORMERS_VERBOSITY"					,  UpperCamelCase       )
							if env_level_str:
														if env_level_str in log_levels:
																					return log_levels[env_level_str]
														else:
																					logging.getLogger().warning(
																					    F"Unknown option TRANSFORMERS_VERBOSITY={env_level_str}, "
																					    F"has to be one of: { ', '.join(log_levels.keys()       ) }"       )
							return _default_log_level






def 					A__				(			):
							return __name__.split("."       )[0]






def 					A__				(			):
							return logging.getLogger(_get_library_name()       )






def 					A__				(			):
							global _default_handler

							with _lock:
														if _default_handler:
																					# This library has already configured the library root logger.
																					return
														A		=  logging.StreamHandler()  # Set sys.stderr as stream.
														A		=  sys.stderr.flush

														# Apply our default configuration to the library root logger.
														A		=  _get_library_root_logger()
														library_root_logger.addHandler(_default_handler       )
														library_root_logger.setLevel(_get_default_logging_level()       )
														A		=  False






def 					A__				(			):
							global _default_handler

							with _lock:
														if not _default_handler:
																					return

														A		=  _get_library_root_logger()
														library_root_logger.removeHandler(_default_handler       )
														library_root_logger.setLevel(logging.NOTSET       )
														A		=  None






def 					A__				(			):
							return log_levels






def 					A__				(			UpperCamelCase = None       ):

							if name is None:
														A		=  _get_library_name()

							_configure_library_root_logger()
							return logging.getLogger(UpperCamelCase       )






def 					A__				(			):
							_configure_library_root_logger()
							return _get_library_root_logger().getEffectiveLevel()






def 					A__				(			UpperCamelCase       ):
							_configure_library_root_logger()
							_get_library_root_logger().setLevel(UpperCamelCase       )






def 					A__				(			):
							return set_verbosity(UpperCamelCase       )






def 					A__				(			):
							return set_verbosity(UpperCamelCase       )






def 					A__				(			):
							return set_verbosity(UpperCamelCase       )






def 					A__				(			):
							return set_verbosity(UpperCamelCase       )






def 					A__				(			):
							_configure_library_root_logger()

							assert _default_handler is not None
							_get_library_root_logger().removeHandler(_default_handler       )






def 					A__				(			):
							_configure_library_root_logger()

							assert _default_handler is not None
							_get_library_root_logger().addHandler(_default_handler       )






def 					A__				(			UpperCamelCase       ):
							_configure_library_root_logger()

							assert handler is not None
							_get_library_root_logger().addHandler(UpperCamelCase       )






def 					A__				(			UpperCamelCase       ):
							_configure_library_root_logger()

							assert handler is not None and handler not in _get_library_root_logger().handlers
							_get_library_root_logger().removeHandler(UpperCamelCase       )






def 					A__				(			):
							_configure_library_root_logger()
							A		=  False






def 					A__				(			):
							_configure_library_root_logger()
							A		=  True






def 					A__				(			):
							A		=  _get_library_root_logger().handlers

							for handler in handlers:
														A		=  logging.Formatter("[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s"       )
														handler.setFormatter(UpperCamelCase       )






def 					A__				(			):
							A		=  _get_library_root_logger().handlers

							for handler in handlers:
														handler.setFormatter(UpperCamelCase       )






def 					A__				(			self					,  *UpperCamelCase					,  **UpperCamelCase       ):
							A		=  os.getenv("TRANSFORMERS_NO_ADVISORY_WARNINGS"					,  UpperCamelCase       )
							if no_advisory_warnings:
														return
							self.warning(*UpperCamelCase					,  **UpperCamelCase       )


_snake_case     :		int       						= warning_advice






@functools.lru_cache(UpperCamelCase       )
def 					A__				(			self					,  *UpperCamelCase					,  **UpperCamelCase       ):
							self.warning(*UpperCamelCase					,  **UpperCamelCase       )


_snake_case     :		Optional[int]       						= warning_once






class 					_UpperCAmelCase   :


					def __init__(					self			:Tuple     ,       *__UpperCamelCase			:str     ,       **__UpperCamelCase			:Optional[Any]		):  # pylint: disable=unused-argument
												A		=  args[0] if args else None


					def __iter__(					self			:str		):
												return iter(self._iterator		)


					def __getattr__(					self			:Optional[int]     ,       __UpperCamelCase			:List[str]		):

												def empty_fn(*__UpperCamelCase			:List[Any]     ,       **__UpperCamelCase			:Optional[int]		):  # pylint: disable=unused-argument
																			return

												return empty_fn


					def __enter__(					self			:Union[str, Any]		):
												return self


					def __exit__(					self			:str     ,       __UpperCamelCase			:List[Any]     ,       __UpperCamelCase			:Optional[Any]     ,       __UpperCamelCase			:int		):
												return







class 					_UpperCAmelCase   :


					def __call__(					self			:str     ,       *__UpperCamelCase			:Any     ,       **__UpperCamelCase			:Dict		):
												if _tqdm_active:
																			return tqdm_lib.tqdm(*__UpperCamelCase     ,       **__UpperCamelCase		)
												else:
																			return EmptyTqdm(*__UpperCamelCase     ,       **__UpperCamelCase		)


					def 	lowerCamelCase       (					self			:Any     ,       *__UpperCamelCase			:Optional[Any]     ,       **__UpperCamelCase			:int		):
												A		=  None
												if _tqdm_active:
																			return tqdm_lib.tqdm.set_lock(*__UpperCamelCase     ,       **__UpperCamelCase		)


					def 	lowerCamelCase       (					self			:List[str]		):
												if _tqdm_active:
																			return tqdm_lib.tqdm.get_lock()







_snake_case     :		Any       						= _tqdm_cls()






def 					A__				(			):
							global _tqdm_active
							return bool(_tqdm_active       )






def 					A__				(			):
							global _tqdm_active
							A		=  True
							hf_hub_utils.enable_progress_bars()






def 					A__				(			):
							global _tqdm_active
							A		=  False
							hf_hub_utils.disable_progress_bars()









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


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

_snake_case     :		List[Any]       						= '\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 					A__				(			UpperCamelCase					,  UpperCamelCase					,  UpperCamelCase=8       ):
							A		=  height // scale_factor**2
							if height % scale_factor**2 != 0:
														new_height += 1
							A		=  width // scale_factor**2
							if width % scale_factor**2 != 0:
														new_width += 1
							return new_height * scale_factor, new_width * scale_factor






class 					_UpperCAmelCase   (	lowercase_ ):


					def __init__(					self			:Any     ,       __UpperCamelCase			:UNetaDConditionModel     ,       __UpperCamelCase			:DDPMScheduler     ,       __UpperCamelCase			:VQModel     ,       ):
												super().__init__()

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


					def 	lowerCamelCase       (					self			:Union[str, Any]     ,       __UpperCamelCase			:Tuple     ,       __UpperCamelCase			:Dict     ,       __UpperCamelCase			:Dict     ,       __UpperCamelCase			:List[str]     ,       __UpperCamelCase			:Optional[int]     ,       __UpperCamelCase			:List[str]		):
												if latents is None:
																			A		=  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}"		)
																			A		=  latents.to(__UpperCamelCase		)

												A		=  latents * scheduler.init_noise_sigma
												return latents


					def 	lowerCamelCase       (					self			:Tuple     ,       __UpperCamelCase			:Any=0		):
												if is_accelerate_available():
																			from accelerate import cpu_offload
												else:
																			raise ImportError("Please install accelerate via `pip install accelerate`"		)

												A		=  torch.device(f"cuda:{gpu_id}"		)

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


					def 	lowerCamelCase       (					self			:Dict     ,       __UpperCamelCase			:int=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."		)

												A		=  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)

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

												# 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			:str		):
												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			:List[Any]     ,       __UpperCamelCase			:Union[torch.FloatTensor, List[torch.FloatTensor]]     ,       __UpperCamelCase			:Union[torch.FloatTensor, List[torch.FloatTensor]]     ,       __UpperCamelCase			:torch.FloatTensor     ,       __UpperCamelCase			:int = 5_12     ,       __UpperCamelCase			:int = 5_12     ,       __UpperCamelCase			:int = 1_00     ,       __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     ,       ):
												A		=  self._execution_device

												A		=  guidance_scale > 1.0

												if isinstance(__UpperCamelCase     ,       __UpperCamelCase		):
																			A		=  torch.cat(__UpperCamelCase     ,       dim=0		)
												if isinstance(__UpperCamelCase     ,       __UpperCamelCase		):
																			A		=  torch.cat(__UpperCamelCase     ,       dim=0		)
												if isinstance(__UpperCamelCase     ,       __UpperCamelCase		):
																			A		=  torch.cat(__UpperCamelCase     ,       dim=0		)

												A		=  image_embeds.shape[0] * num_images_per_prompt

												if do_classifier_free_guidance:
																			A		=  image_embeds.repeat_interleave(__UpperCamelCase     ,       dim=0		)
																			A		=  negative_image_embeds.repeat_interleave(__UpperCamelCase     ,       dim=0		)
																			A		=  hint.repeat_interleave(__UpperCamelCase     ,       dim=0		)

												A		=  torch.cat([negative_image_embeds, image_embeds]     ,       dim=0		).to(dtype=self.unet.dtype     ,       device=__UpperCamelCase		)
												A		=  torch.cat([hint, hint]     ,       dim=0		).to(dtype=self.unet.dtype     ,       device=__UpperCamelCase		)
												self.scheduler.set_timesteps(__UpperCamelCase     ,       device=__UpperCamelCase		)
												A		=  self.scheduler.timesteps

												A		=  self.movq.config.latent_channels

												A,	A		=  downscale_height_and_width(__UpperCamelCase     ,       __UpperCamelCase     ,       self.movq_scale_factor		)

												# create initial latent
												A		=  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
																			A		=  torch.cat([latents] * 2		) if do_classifier_free_guidance else latents

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

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

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

												if output_type == "pil":
																			A		=  self.numpy_to_pil(__UpperCamelCase		)

												if not return_dict:
																			return (image,)

												return ImagePipelineOutput(images=__UpperCamelCase		)


