


from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

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:						Union[str, Any]   				=       logging.get_logger(__name__)  # pylint: disable=invalid-name

_lowerCAmelCase:						Optional[Any]   				=       '\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Img2ImgPipeline, KandinskyV22PriorPipeline\n        >>> from diffusers.utils import load_image\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior.to("cuda")\n\n        >>> prompt = "A red cartoon frog, 4k"\n        >>> image_emb, zero_image_emb = pipe_prior(prompt, return_dict=False)\n\n        >>> pipe = KandinskyV22Img2ImgPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-decoder", torch_dtype=torch.float16\n        ... )\n        >>> pipe.to("cuda")\n\n        >>> init_image = load_image(\n        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"\n        ...     "/kandinsky/frog.png"\n        ... )\n\n        >>> image = pipe(\n        ...     image=init_image,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ...     strength=0.2,\n        ... ).images\n\n        >>> image[0].save("red_frog.png")\n        ```\n'






def   _lowercase(					__a				:						Any    ,     __a				:						int    ,     __a				:						str=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






def   _lowercase(					__a				:						Tuple    ,     __a				:						Optional[Any]=512    ,     __a				:						Any=512     ):
							a__      =pil_image.resize((w, h)    ,     resample=Image.BICUBIC    ,     reducing_gap=1     )
							a__      =np.array(pil_image.convert('RGB'     )     )
							a__      =arr.astype(np.floataa     ) / 1_27.5 - 1
							a__      =np.transpose(__a    ,     [2, 0, 1]     )
							a__      =torch.from_numpy(__a     ).unsqueeze(0     )
							return image







class       lowercase_					(lowercase__   ):





				def __init__(				self       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  )  -> Any:
											super().__init__()

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





				def 						__UpperCamelCase    (				self       ,  lowercase_       ,  lowercase_       ,  lowercase_)  -> str:
											# get the original timestep using init_timestep
											a__      =min(int(num_inference_steps * strength)       ,  lowercase_)

											a__      =max(num_inference_steps - init_timestep       ,  0)
											a__      =self.scheduler.timesteps[t_start:]

											return timesteps, num_inference_steps - t_start





				def 						__UpperCamelCase    (				self       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_=None)  -> Union[str, Any]:
											if not isinstance(lowercase_       ,  (torch.Tensor, PIL.Image.Image, list)):
																		raise ValueError(
																		    F"""`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(lowercase_)}""")

											a__      =image.to(device=lowercase_       ,  dtype=lowercase_)

											a__      =batch_size * num_images_per_prompt

											if image.shape[1] == 4:
																		a__      =image

											else:
																		if isinstance(lowercase_       ,  lowercase_) and len(lowercase_) != batch_size:
																									raise ValueError(
																									    F"""You have passed a list of generators of length {len(lowercase_)}, but requested an effective batch"""
																									    F""" size of {batch_size}. Make sure the batch size matches the length of the generators.""")

																		elif isinstance(lowercase_       ,  lowercase_):
																									a__      =[
																									    self.movq.encode(image[i : i + 1]).latent_dist.sample(generator[i]) for i in range(lowercase_)
																									]
																									a__      =torch.cat(lowercase_       ,  dim=0)
																		else:
																									a__      =self.movq.encode(lowercase_).latent_dist.sample(lowercase_)

																		a__      =self.movq.config.scaling_factor * init_latents

											a__      =torch.cat([init_latents]       ,  dim=0)

											a__      =init_latents.shape
											a__      =randn_tensor(lowercase_       ,  generator=lowercase_       ,  device=lowercase_       ,  dtype=lowercase_)

											# get latents
											a__      =self.scheduler.add_noise(lowercase_       ,  lowercase_       ,  lowercase_)

											a__      =init_latents

											return latents





				def 						__UpperCamelCase    (				self       ,  lowercase_=0)  -> str:
											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(lowercase_       ,  lowercase_)





				def 						__UpperCamelCase    (				self       ,  lowercase_=0)  -> List[Any]:
											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=lowercase_)
																		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(lowercase_       ,  lowercase_       ,  prev_module_hook=lowercase_)

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





				@property
				# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
				def 						__UpperCamelCase    (				self)  -> List[Any]:
											if not hasattr(self.unet       ,  '_hf_hook'):
																		return self.device
											for module in self.unet.modules():
																		if (
																		    hasattr(lowercase_       ,  '_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(lowercase_)
				def __call__(				self       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_ = 512       ,  lowercase_ = 512       ,  lowercase_ = 100       ,  lowercase_ = 4.0       ,  lowercase_ = 0.3       ,  lowercase_ = 1       ,  lowercase_ = None       ,  lowercase_ = "pil"       ,  lowercase_ = True       ,  )  -> Optional[int]:
											a__      =self._execution_device

											a__      =guidance_scale > 1.0

											if isinstance(lowercase_       ,  lowercase_):
																		a__      =torch.cat(lowercase_       ,  dim=0)
											a__      =image_embeds.shape[0]
											if isinstance(lowercase_       ,  lowercase_):
																		a__      =torch.cat(lowercase_       ,  dim=0)

											if do_classifier_free_guidance:
																		a__      =image_embeds.repeat_interleave(lowercase_       ,  dim=0)
																		a__      =negative_image_embeds.repeat_interleave(lowercase_       ,  dim=0)

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

											if not isinstance(lowercase_       ,  lowercase_):
																		a__      =[image]
											if not all(isinstance(lowercase_       ,  (PIL.Image.Image, torch.Tensor)) for i in image):
																		raise ValueError(
																		    F"""Input is in incorrect format: {[type(lowercase_) for i in image]}. Currently, we only support  PIL image and pytorch tensor""")

											a__      =torch.cat([prepare_image(lowercase_       ,  lowercase_       ,  lowercase_) for i in image]       ,  dim=0)
											a__      =image.to(dtype=image_embeds.dtype       ,  device=lowercase_)

											a__      =self.movq.encode(lowercase_)['latents']
											a__      =latents.repeat_interleave(lowercase_       ,  dim=0)
											self.scheduler.set_timesteps(lowercase_       ,  device=lowercase_)
											a__    ,     a__      =self.get_timesteps(lowercase_       ,  lowercase_       ,  lowercase_)
											a__      =timesteps[:1].repeat(batch_size * num_images_per_prompt)
											a__    ,     a__      =downscale_height_and_width(lowercase_       ,  lowercase_       ,  self.movq_scale_factor)
											a__      =self.prepare_latents(
											    lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  image_embeds.dtype       ,  lowercase_       ,  lowercase_)
											for i, t in enumerate(self.progress_bar(lowercase_)):
																		# 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}
																		a__      =self.unet(
																		    sample=lowercase_       ,  timestep=lowercase_       ,  encoder_hidden_states=lowercase_       ,  added_cond_kwargs=lowercase_       ,  return_dict=lowercase_       ,  )[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(
																		    lowercase_       ,  lowercase_       ,  lowercase_       ,  generator=lowercase_       ,  )[0]

											# post-processing
											a__      =self.movq.decode(lowercase_       ,  force_not_quantize=lowercase_)['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(lowercase_)

											if not return_dict:
																		return (image,)

											return ImagePipelineOutput(images=lowercase_)








'''simple docstring'''


from ...configuration_utils import PretrainedConfig
from ...utils import logging


_SCREAMING_SNAKE_CASE									=				logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE									=				{
    "tiiuae/falcon-40b": "https://huggingface.co/tiiuae/falcon-40b/resolve/main/config.json",
    "tiiuae/falcon-7b": "https://huggingface.co/tiiuae/falcon-7b/resolve/main/config.json",
}

class  lowerCAmelCase_		(		__magic_name__		):
 __lowerCamelCase    :				Any								=       "falcon"
 __lowerCamelCase    :				List[str]								=       ["past_key_values"]




 def __init__(						self      ,  _lowerCAmelCase=65024      ,  _lowerCAmelCase=4544      ,  _lowerCAmelCase=32      ,  _lowerCAmelCase=71      ,  _lowerCAmelCase=1E-5      ,  _lowerCAmelCase=0.02      ,  _lowerCAmelCase=True      ,  _lowerCAmelCase=0.0      ,  _lowerCAmelCase=0.0      ,  _lowerCAmelCase=None      ,  _lowerCAmelCase=False      ,  _lowerCAmelCase=False      ,  _lowerCAmelCase=True      ,  _lowerCAmelCase=True      ,  _lowerCAmelCase=False      ,  _lowerCAmelCase=11      ,  _lowerCAmelCase=11      ,  **_lowerCAmelCase      ,  )     ->				Union[str, Any]:
     _lowerCAmelCase						      =   vocab_size
     # Backward compatibility with n_embed kwarg
     _lowerCAmelCase						      =   kwargs.pop("n_embed"      ,  _lowerCAmelCase		)
     _lowerCAmelCase						      =   hidden_size if n_embed is None else n_embed
     _lowerCAmelCase						      =   num_hidden_layers
     _lowerCAmelCase						      =   num_attention_heads
     _lowerCAmelCase						      =   layer_norm_epsilon
     _lowerCAmelCase						      =   initializer_range
     _lowerCAmelCase						      =   use_cache
     _lowerCAmelCase						      =   hidden_dropout
     _lowerCAmelCase						      =   attention_dropout

     _lowerCAmelCase						      =   bos_token_id
     _lowerCAmelCase						      =   eos_token_id
     _lowerCAmelCase						      =   num_attention_heads if num_kv_heads is None else num_kv_heads
     _lowerCAmelCase						      =   alibi
     _lowerCAmelCase						      =   new_decoder_architecture
     _lowerCAmelCase						      =   multi_query  # Ignored when new_decoder_architecture is True
     _lowerCAmelCase						      =   parallel_attn
     _lowerCAmelCase						      =   bias

     super().__init__(bos_token_id=_lowerCAmelCase      ,  eos_token_id=_lowerCAmelCase      ,  **_lowerCAmelCase		)




 @property
 def 				_snake_case (						self		)     ->				Optional[Any]:
     return self.hidden_size // self.num_attention_heads





 @property
 def 				_snake_case (						self		)     ->				Optional[Any]:
     return not self.alibi


