

from typing import Callable, List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModel, CLIPTokenizer

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin, TransformeraDModel, VQModel
from ...schedulers import VQDiffusionScheduler
from ...utils import logging
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


A_	:			List[Any]  	=      logging.get_logger(__name__)  # pylint: disable=invalid-name



class    _a						(__magic_name__		,       __magic_name__    ):


					'''simple docstring'''






					@register_to_config
					def __init__(			self   ,	A__   ,	A__ = None   ,	A__ = None      ):
						super().__init__()

						A__							:   int					       =		learnable

						if self.learnable:
							assert hidden_size is not None, "learnable=True requires `hidden_size` to be set"
							assert length is not None, "learnable=True requires `length` to be set"

							A__							:   List[str]					       =		torch.zeros(A__   ,	A__      )
						else:
							A__							:   str					       =		None

						A__							:   Optional[int]					       =		torch.nn.Parameter(A__      )



class    _a						(__magic_name__    ):


					'''simple docstring'''






					UpperCAmelCase__:       VQModel
					UpperCAmelCase__:       CLIPTextModel
					UpperCAmelCase__:       CLIPTokenizer
					UpperCAmelCase__:       TransformeraDModel
					UpperCAmelCase__:       LearnedClassifierFreeSamplingEmbeddings
					UpperCAmelCase__:       VQDiffusionScheduler






					def __init__(			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	):
						super().__init__()

						self.register_modules(
						    vqvae=A__   ,	transformer=A__   ,	text_encoder=A__   ,	tokenizer=A__   ,	scheduler=A__   ,	learned_classifier_free_sampling_embeddings=A__   ,	)






					def   __A      (			self   ,	A__   ,	A__   ,	A__      ):
						A__							:   Optional[int]					       =		len(A__      ) if isinstance(A__   ,	A__      ) else 1

						# get prompt text embeddings
						A__							:   int					       =		self.tokenizer(
						    A__   ,	padding="""max_length"""   ,	max_length=self.tokenizer.model_max_length   ,	return_tensors="""pt"""   ,	)
						A__							:   Tuple					       =		text_inputs.input_ids

						if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
							A__							:   List[Any]					       =		self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]      )
							logger.warning(
							    """The following part of your input was truncated because CLIP can only handle sequences up to"""
							    F""" {self.tokenizer.model_max_length} tokens: {removed_text}"""      )
							A__							:   Dict					       =		text_input_ids[:, : self.tokenizer.model_max_length]
						A__							:   int					       =		self.text_encoder(text_input_ids.to(self.device      )      )[0]

						# NOTE: This additional step of normalizing the text embeddings is from VQ-Diffusion.
						# While CLIP does normalize the pooled output of the text transformer when combining
						# the image and text embeddings, CLIP does not directly normalize the last hidden state.
						#
						# CLIP normalizing the pooled output.
						# https://github.com/huggingface/transformers/blob/d92e22d1f28324f513f3080e5c47c071a3916721/src/transformers/models/clip/modeling_clip.py#L1052-L1053
						A__							:   Union[str, Any]					       =		prompt_embeds / prompt_embeds.norm(dim=-1   ,	keepdim=A__      )

						# duplicate text embeddings for each generation per prompt
						A__							:   Union[str, Any]					       =		prompt_embeds.repeat_interleave(A__   ,	dim=0      )

						if do_classifier_free_guidance:
							if self.learned_classifier_free_sampling_embeddings.learnable:
								A__							:   Any					       =		self.learned_classifier_free_sampling_embeddings.embeddings
								A__							:   int					       =		negative_prompt_embeds.unsqueeze(0      ).repeat(A__   ,	1   ,	1      )
							else:
								A__							:   Optional[Any]					       =		[""""""] * batch_size

								A__							:   Optional[int]					       =		text_input_ids.shape[-1]
								A__							:   int					       =		self.tokenizer(
								    A__   ,	padding="""max_length"""   ,	max_length=A__   ,	truncation=A__   ,	return_tensors="""pt"""   ,	)
								A__							:   Dict					       =		self.text_encoder(uncond_input.input_ids.to(self.device      )      )[0]
								# See comment for normalizing text embeddings
								A__							:   Dict					       =		negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1   ,	keepdim=A__      )

							# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
							A__							:   Union[str, Any]					       =		negative_prompt_embeds.shape[1]
							A__							:   Union[str, Any]					       =		negative_prompt_embeds.repeat(1   ,	A__   ,	1      )
							A__							:   Any					       =		negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,	A__   ,	-1      )

							# For classifier free guidance, we need to do two forward passes.
							# Here we concatenate the unconditional and text embeddings into a single batch
							# to avoid doing two forward passes
							A__							:   Optional[int]					       =		torch.cat([negative_prompt_embeds, prompt_embeds]      )

						return prompt_embeds






					@torch.no_grad()
					def __call__(			self   ,	A__   ,	A__ = 100   ,	A__ = 5.0   ,	A__ = 1.0   ,	A__ = 1   ,	A__ = None   ,	A__ = None   ,	A__ = "pil"   ,	A__ = True   ,	A__ = None   ,	A__ = 1   ,	):
						if isinstance(A__   ,	A__      ):
							A__							:   Any					       =		1
						elif isinstance(A__   ,	A__      ):
							A__							:   Optional[Any]					       =		len(A__      )
						else:
							raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(A__      )}"""      )

						A__							:   Tuple					       =		batch_size * num_images_per_prompt

						A__							:   List[str]					       =		guidance_scale > 1.0

						A__							:   List[str]					       =		self._encode_prompt(A__   ,	A__   ,	A__      )

						if (callback_steps is None) or (
						    callback_steps is not None and (not isinstance(A__   ,	A__      ) or callback_steps <= 0)
						):
							raise ValueError(
							    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
							    F""" {type(A__      )}."""      )

						# get the initial completely masked latents unless the user supplied it

						A__							:   List[str]					       =		(batch_size, self.transformer.num_latent_pixels)
						if latents is None:
							A__							:   int					       =		self.transformer.num_vector_embeds - 1
							A__							:   Tuple					       =		torch.full(A__   ,	A__      ).to(self.device      )
						else:
							if latents.shape != latents_shape:
								raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""      )
							if (latents < 0).any() or (latents >= self.transformer.num_vector_embeds).any():
								raise ValueError(
								    """Unexpected latents value(s). All latents be valid embedding indices i.e. in the range 0,"""
								    F""" {self.transformer.num_vector_embeds - 1} (inclusive)."""      )
							A__							:   List[str]					       =		latents.to(self.device      )

						# set timesteps
						self.scheduler.set_timesteps(A__   ,	device=self.device      )

						A__							:   Any					       =		self.scheduler.timesteps.to(self.device      )

						A__							:   Tuple					       =		latents

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

							# predict the un-noised image
							# model_output == `log_p_x_0`
							A__							:   Optional[Any]					       =		self.transformer(A__   ,	encoder_hidden_states=A__   ,	timestep=A__      ).sample

							if do_classifier_free_guidance:
								A__       ,       A__							:   int					       =		model_output.chunk(2      )
								A__							:   str					       =		model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
								model_output -= torch.logsumexp(A__   ,	dim=1   ,	keepdim=A__      )

							A__							:   Optional[Any]					       =		self.truncate(A__   ,	A__      )

							# remove `log(0)`'s (`-inf`s)
							A__							:   str					       =		model_output.clamp(-70      )

							# compute the previous noisy sample x_t -> x_t-1
							A__							:   int					       =		self.scheduler.step(A__   ,	timestep=A__   ,	sample=A__   ,	generator=A__      ).prev_sample

							# call the callback, if provided
							if callback is not None and i % callback_steps == 0:
								callback(A__   ,	A__   ,	A__      )

						A__							:   Any					       =		self.vqvae.config.vq_embed_dim
						A__							:   int					       =		(batch_size, self.transformer.height, self.transformer.width, embedding_channels)
						A__							:   str					       =		self.vqvae.quantize.get_codebook_entry(A__   ,	shape=A__      )
						A__							:   Tuple					       =		self.vqvae.decode(A__   ,	force_not_quantize=A__      ).sample

						A__							:   Dict					       =		(image / 2 + 0.5).clamp(0   ,	1      )
						A__							:   List[Any]					       =		image.cpu().permute(0   ,	2   ,	3   ,	1      ).numpy()

						if output_type == "pil":
							A__							:   int					       =		self.numpy_to_pil(A__      )

						if not return_dict:
							return (image,)

						return ImagePipelineOutput(images=A__      )






					def   __A      (			self   ,	A__   ,	A__      ):
						A__       ,       A__							:   Union[str, Any]					       =		torch.sort(A__   ,	1   ,	descending=A__      )
						A__							:   List[Any]					       =		torch.exp(A__      )
						A__							:   Optional[int]					       =		sorted_p_x_0.cumsum(dim=1      ) < truncation_rate

						# Ensure that at least the largest probability is not zeroed out
						A__							:   List[str]					       =		torch.full_like(keep_mask[:, 0:1, :]   ,	A__      )
						A__							:   List[Any]					       =		torch.cat((all_true, keep_mask)   ,	dim=1      )
						A__							:   List[str]					       =		keep_mask[:, :-1, :]

						A__							:   Dict					       =		keep_mask.gather(1   ,	indices.argsort(1      )      )

						A__							:   Optional[Any]					       =		log_p_x_0.clone()

						A__							:   int					       =		-torch.inf  # -inf = log(0)

						return rv






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


A_	:			str  	=      logging.get_logger(__name__)

A_	:			Any  	=      {
    '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    _a						(__magic_name__    ):


					'''simple docstring'''






					UpperCAmelCase__:       Dict  							=	'''falcon'''
					UpperCAmelCase__:       Any  							=	['''past_key_values''']






					def __init__(			self   ,	A__=6_5024   ,	A__=4544   ,	A__=32   ,	A__=71   ,	A__=1e-5   ,	A__=0.0_2   ,	A__=True   ,	A__=0.0   ,	A__=0.0   ,	A__=None   ,	A__=False   ,	A__=False   ,	A__=True   ,	A__=True   ,	A__=False   ,	A__=11   ,	A__=11   ,	**A__   ,	):
						A__							:   Dict					       =		vocab_size
						# Backward compatibility with n_embed kwarg
						A__							:   Union[str, Any]					       =		kwargs.pop("""n_embed"""   ,	A__      )
						A__							:   Optional[Any]					       =		hidden_size if n_embed is None else n_embed
						A__							:   Any					       =		num_hidden_layers
						A__							:   Any					       =		num_attention_heads
						A__							:   Optional[Any]					       =		layer_norm_epsilon
						A__							:   Tuple					       =		initializer_range
						A__							:   Tuple					       =		use_cache
						A__							:   str					       =		hidden_dropout
						A__							:   List[str]					       =		attention_dropout

						A__							:   List[Any]					       =		bos_token_id
						A__							:   Optional[Any]					       =		eos_token_id
						A__							:   Optional[Any]					       =		num_attention_heads if num_kv_heads is None else num_kv_heads
						A__							:   List[str]					       =		alibi
						A__							:   Tuple					       =		new_decoder_architecture
						A__							:   List[str]					       =		multi_query  # Ignored when new_decoder_architecture is True
						A__							:   List[Any]					       =		parallel_attn
						A__							:   int					       =		bias

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






					@property
					def   __A      (			self      ):
						return self.hidden_size // self.num_attention_heads






					@property
					def   __A      (			self      ):
						return not self.alibi



