





import os
from shutil import copyfile
from typing import TYPE_CHECKING, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging
from ...utils.versions import require_version


if TYPE_CHECKING:
				from transformers.pipelines.conversational import Conversation

require_version('''tokenizers>=0.13.3''')

if is_sentencepiece_available():
				from .tokenization_llama import LlamaTokenizer
else:
				snake_case_				=      None

snake_case_				=      logging.get_logger(__name__)
snake_case_				=      {'''vocab_file''': '''tokenizer.model''', '''tokenizer_file''': '''tokenizer.json'''}

snake_case_,							snake_case_				=      '''[INST]''', '''[/INST]'''
snake_case_,							snake_case_				=      '''<<SYS>>\n''', '''\n<</SYS>>\n\n'''

# fmt: off
snake_case_				=      '''You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\
answers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\
that your responses are socially unbiased and positive in nature.

If a question does not make any sense, or is not factually coherent, explain why instead of answering something not\
correct. If you don\'t know the answer to a question, please don\'t share false information.'''
class 							snake_case_		(      _A):
		lowerCamelCase     :Optional[Any]							=  VOCAB_FILES_NAMES
		lowerCamelCase     :Union[str, Any]							=  LlamaTokenizer
		lowerCamelCase     :Dict							=  "left"
		lowerCamelCase     :int							=  ["input_ids", "attention_mask"]







		def __init__(		self      ,  __lowercase=None      ,  __lowercase=None      ,  __lowercase=False      ,  __lowercase="<unk>"      ,  __lowercase="<s>"      ,  __lowercase="</s>"      ,  __lowercase=True      ,  __lowercase=False      ,  **__lowercase      ,  )					-> Union[str, Any]:
				super().__init__(
				    vocab_file=__lowercase      ,  tokenizer_file=__lowercase      ,  clean_up_tokenization_spaces=__lowercase      ,  unk_token=__lowercase      ,  bos_token=__lowercase      ,  eos_token=__lowercase      ,  **__lowercase      ,  )
				lowerCamelCase :			str    		=add_bos_token
				lowerCamelCase :			Any    		=add_eos_token
				self.update_post_processor()

				lowerCamelCase :			List[Any]    		=vocab_file
				lowerCamelCase :			Optional[int]    		=False if not self.vocab_file else True







		def       __lowercase (		self )					-> Optional[int]:
				lowerCamelCase :			Any    		=self.bos_token
				lowerCamelCase :			Any    		=self.bos_token_id

				lowerCamelCase :			Tuple    		=self.eos_token
				lowerCamelCase :			Dict    		=self.eos_token_id

				lowerCamelCase :			List[str]    		=F"{(bos+':0 ') * self.add_bos_token}$A:0{(' '+eos+':0') * self.add_eos_token}"
				lowerCamelCase :			List[str]    		=F"{single}{(' '+bos+':1') * self.add_bos_token} $B:1{(' '+eos+':1') * self.add_eos_token}"

				lowerCamelCase :			Dict    		=[]
				if self.add_bos_token:
						special_tokens.append((bos, bos_token_id) )
				if self.add_eos_token:
						special_tokens.append((eos, eos_token_id) )
				lowerCamelCase :			Optional[int]    		=processors.TemplateProcessing(
				    single=__lowercase      ,  pair=__lowercase      ,  special_tokens=__lowercase )







		@property
		def       __lowercase (		self )					-> Any:
				return self._add_eos_token







		@property
		def       __lowercase (		self )					-> int:
				return self._add_bos_token







		@add_eos_token.setter
		def       __lowercase (		self      ,  __lowercase )					-> Any:
				lowerCamelCase :			Union[str, Any]    		=value
				self.update_post_processor()







		@add_bos_token.setter
		def       __lowercase (		self      ,  __lowercase )					-> Optional[Any]:
				lowerCamelCase :			Optional[Any]    		=value
				self.update_post_processor()







		def       __lowercase (		self      ,  __lowercase      ,  __lowercase = None )					-> Tuple[str]:
				if not self.can_save_slow_tokenizer:
						raise ValueError(
						    '''Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '''
						    '''tokenizer.''' )

				if not os.path.isdir(__lowercase ):
						logger.error(F"Vocabulary path ({save_directory}) should be a directory" )
						return
				lowerCamelCase :			str    		=os.path.join(
				    __lowercase      ,  (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file'''] )

				if os.path.abspath(self.vocab_file ) != os.path.abspath(__lowercase ):
						copyfile(self.vocab_file      ,  __lowercase )

				return (out_vocab_file,)







		def       __lowercase (		self      ,  __lowercase )					-> Union[str, Any]:
				lowerCamelCase :			Union[str, Any]    		=list(conversation.iter_texts() )
				if not all([is_user for is_user, msg in dialogue[::2]] ) or not all(
				    [not is_user for is_user, msg in dialogue[1::2]] ):
						raise ValueError(
						    '''The model only supports \'user\' and \'assistant\' roles, starting with user and alternating (u/a/u/a/u...)''' )

				lowerCamelCase :			int    		=[]
				if len(conversation.past_user_inputs ) > 0:
						if not conversation.past_user_inputs[0].startswith(__lowercase ) or E_SYS not in conversation.past_user_inputs[0]:
								lowerCamelCase :			Union[str, Any]    		=(
								    B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
								)
				elif not dialogue[0][1].startswith(__lowercase ) or E_SYS not in dialogue[0][1]:
						lowerCamelCase :			Optional[Any]    		=(dialogue[0][0], B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + dialogue[0][1])

				dialog_tokens += sum(
				    [
				        [self.bos_token_id]
				        + self.encode(
				            F"{B_INST} {(prompt[1]).strip()} {E_INST} {(answer[1]).strip()} "      ,  add_special_tokens=__lowercase )
				        + [self.eos_token_id]
				        for prompt, answer in zip(dialogue[::2]      ,  dialogue[1::2] )
				    ]      ,  []      ,  )
				if not (dialogue[-1][0]):
						raise ValueError(F"Last message must be from user, got {dialogue[-1]['role']}" )
				dialog_tokens += [self.bos_token_id] + self.encode(
				    F"{B_INST} {(dialogue[-1][1]).strip()} {E_INST}"      ,  add_special_tokens=__lowercase )
				return dialog_tokens








import inspect
from typing import List, Optional, Union

import torch
from torch import nn
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPImageProcessor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
class 							snake_case_		(      nn.Module):







		def __init__(		self      ,  __lowercase      ,  __lowercase=1.0 )					-> List[str]:
				super().__init__()

				lowerCamelCase :			Optional[Any]    		=cut_size
				lowerCamelCase :			List[str]    		=cut_power







		def       __lowercase (		self      ,  __lowercase      ,  __lowercase )					-> int:
				lowerCamelCase		, lowerCamelCase :			Any    		=pixel_values.shape[2:4]
				lowerCamelCase :			Any    		=min(__lowercase      ,  __lowercase )
				lowerCamelCase :			int    		=min(__lowercase      ,  __lowercase      ,  self.cut_size )
				lowerCamelCase :			Dict    		=[]
				for _ in range(__lowercase ):
						lowerCamelCase :			Union[str, Any]    		=int(torch.rand([] ) ** self.cut_power * (max_size - min_size) + min_size )
						lowerCamelCase :			Any    		=torch.randint(0      ,  sideX - size + 1      ,  () )
						lowerCamelCase :			Any    		=torch.randint(0      ,  sideY - size + 1      ,  () )
						lowerCamelCase :			Optional[Any]    		=pixel_values[:, :, offsety : offsety + size, offsetx : offsetx + size]
						cutouts.append(F.adaptive_avg_poolad(__lowercase      ,  self.cut_size ) )
				return torch.cat(__lowercase )


def        A__						(							SCREAMING_SNAKE_CASE_				,							SCREAMING_SNAKE_CASE_		) ->  Optional[Any]:
		lowerCamelCase :			str    		=F.normalize(SCREAMING_SNAKE_CASE_				,							dim=-1		)
		lowerCamelCase :			Optional[Any]    		=F.normalize(SCREAMING_SNAKE_CASE_				,							dim=-1		)
		return (x - y).norm(dim=-1		).div(2		).arcsin().pow(2		).mul(2		)


def        A__						(							SCREAMING_SNAKE_CASE_				,							SCREAMING_SNAKE_CASE_		) ->  List[str]:
		for param in model.parameters():
				lowerCamelCase :			List[Any]    		=value



class 							snake_case_		(      _A):







		def __init__(		self      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  )					-> List[Any]:
				super().__init__()
				self.register_modules(
				    vae=__lowercase      ,  text_encoder=__lowercase      ,  clip_model=__lowercase      ,  tokenizer=__lowercase      ,  unet=__lowercase      ,  scheduler=__lowercase      ,  feature_extractor=__lowercase      ,  )

				lowerCamelCase :			Optional[Any]    		=transforms.Normalize(mean=feature_extractor.image_mean      ,  std=feature_extractor.image_std )
				lowerCamelCase :			int    		=(
				    feature_extractor.size
				    if isinstance(feature_extractor.size      ,  __lowercase )
				    else feature_extractor.size['''shortest_edge''']
				)
				lowerCamelCase :			int    		=MakeCutouts(self.cut_out_size )

				set_requires_grad(self.text_encoder      ,  __lowercase )
				set_requires_grad(self.clip_model      ,  __lowercase )







		def       __lowercase (		self      ,  __lowercase = "auto" )					-> Dict:
				if slice_size == "auto":
						# half the attention head size is usually a good trade-off between
						# speed and memory
						lowerCamelCase :			int    		=self.unet.config.attention_head_dim // 2
				self.unet.set_attention_slice(__lowercase )







		def       __lowercase (		self )					-> List[str]:
				self.enable_attention_slicing(__lowercase )







		def       __lowercase (		self )					-> List[str]:
				set_requires_grad(self.vae      ,  __lowercase )







		def       __lowercase (		self )					-> List[str]:
				set_requires_grad(self.vae      ,  __lowercase )







		def       __lowercase (		self )					-> Dict:
				set_requires_grad(self.unet      ,  __lowercase )







		def       __lowercase (		self )					-> Union[str, Any]:
				set_requires_grad(self.unet      ,  __lowercase )







		@torch.enable_grad()
		def       __lowercase (		self      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase=True      ,  )					-> Union[str, Any]:
				lowerCamelCase :			Dict    		=latents.detach().requires_grad_()

				lowerCamelCase :			int    		=self.scheduler.scale_model_input(__lowercase      ,  __lowercase )

				# predict the noise residual
				lowerCamelCase :			Union[str, Any]    		=self.unet(__lowercase      ,  __lowercase      ,  encoder_hidden_states=__lowercase ).sample

				if isinstance(self.scheduler      ,  (PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler) ):
						lowerCamelCase :			Optional[int]    		=self.scheduler.alphas_cumprod[timestep]
						lowerCamelCase :			List[str]    		=1 - alpha_prod_t
						# compute predicted original sample from predicted noise also called
						# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
						lowerCamelCase :			Tuple    		=(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

						lowerCamelCase :			List[str]    		=torch.sqrt(__lowercase )
						lowerCamelCase :			str    		=pred_original_sample * (fac) + latents * (1 - fac)
				elif isinstance(self.scheduler      ,  __lowercase ):
						lowerCamelCase :			Optional[int]    		=self.scheduler.sigmas[index]
						lowerCamelCase :			str    		=latents - sigma * noise_pred
				else:
						raise ValueError(F"scheduler type {type(self.scheduler )} not supported" )

				lowerCamelCase :			List[Any]    		=1 / self.vae.config.scaling_factor * sample
				lowerCamelCase :			Optional[Any]    		=self.vae.decode(__lowercase ).sample
				lowerCamelCase :			Dict    		=(image / 2 + 0.5).clamp(0      ,  1 )

				if use_cutouts:
						lowerCamelCase :			Tuple    		=self.make_cutouts(__lowercase      ,  __lowercase )
				else:
						lowerCamelCase :			str    		=transforms.Resize(self.cut_out_size )(__lowercase )
				lowerCamelCase :			List[Any]    		=self.normalize(__lowercase ).to(latents.dtype )

				lowerCamelCase :			Optional[Any]    		=self.clip_model.get_image_features(__lowercase )
				lowerCamelCase :			Optional[int]    		=image_embeddings_clip / image_embeddings_clip.norm(p=2      ,  dim=-1      ,  keepdim=__lowercase )

				if use_cutouts:
						lowerCamelCase :			str    		=spherical_dist_loss(__lowercase      ,  __lowercase )
						lowerCamelCase :			Union[str, Any]    		=dists.view([num_cutouts, sample.shape[0], -1] )
						lowerCamelCase :			Tuple    		=dists.sum(2 ).mean(0 ).sum() * clip_guidance_scale
				else:
						lowerCamelCase :			Union[str, Any]    		=spherical_dist_loss(__lowercase      ,  __lowercase ).mean() * clip_guidance_scale

				lowerCamelCase :			List[Any]    		=-torch.autograd.grad(__lowercase      ,  __lowercase )[0]

				if isinstance(self.scheduler      ,  __lowercase ):
						lowerCamelCase :			Any    		=latents.detach() + grads * (sigma**2)
						lowerCamelCase :			Dict    		=noise_pred_original
				else:
						lowerCamelCase :			Union[str, Any]    		=noise_pred_original - torch.sqrt(__lowercase ) * grads
				return noise_pred, latents







		@torch.no_grad()
		def __call__(		self      ,  __lowercase      ,  __lowercase = 5_1_2      ,  __lowercase = 5_1_2      ,  __lowercase = 5_0      ,  __lowercase = 7.5      ,  __lowercase = 1      ,  __lowercase = 0.0      ,  __lowercase = 1_0_0      ,  __lowercase = None      ,  __lowercase = 4      ,  __lowercase = True      ,  __lowercase = None      ,  __lowercase = None      ,  __lowercase = "pil"      ,  __lowercase = True      ,  )					-> Dict:
				if isinstance(__lowercase      ,  __lowercase ):
						lowerCamelCase :			List[str]    		=1
				elif isinstance(__lowercase      ,  __lowercase ):
						lowerCamelCase :			Dict    		=len(__lowercase )
				else:
						raise ValueError(F"`prompt` has to be of type `str` or `list` but is {type(__lowercase )}" )

				if height % 8 != 0 or width % 8 != 0:
						raise ValueError(F"`height` and `width` have to be divisible by 8 but are {height} and {width}." )

				# get prompt text embeddings
				lowerCamelCase :			Optional[int]    		=self.tokenizer(
				    __lowercase      ,  padding='''max_length'''      ,  max_length=self.tokenizer.model_max_length      ,  truncation=__lowercase      ,  return_tensors='''pt'''      ,  )
				lowerCamelCase :			Any    		=self.text_encoder(text_input.input_ids.to(self.device ) )[0]
				# duplicate text embeddings for each generation per prompt
				lowerCamelCase :			Union[str, Any]    		=text_embeddings.repeat_interleave(__lowercase      ,  dim=0 )

				if clip_guidance_scale > 0:
						if clip_prompt is not None:
								lowerCamelCase :			Dict    		=self.tokenizer(
								    __lowercase      ,  padding='''max_length'''      ,  max_length=self.tokenizer.model_max_length      ,  truncation=__lowercase      ,  return_tensors='''pt'''      ,  ).input_ids.to(self.device )
						else:
								lowerCamelCase :			List[Any]    		=text_input.input_ids.to(self.device )
						lowerCamelCase :			Dict    		=self.clip_model.get_text_features(__lowercase )
						lowerCamelCase :			int    		=text_embeddings_clip / text_embeddings_clip.norm(p=2      ,  dim=-1      ,  keepdim=__lowercase )
						# duplicate text embeddings clip for each generation per prompt
						lowerCamelCase :			List[Any]    		=text_embeddings_clip.repeat_interleave(__lowercase      ,  dim=0 )

				# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
				# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
				# corresponds to doing no classifier free guidance.
				lowerCamelCase :			int    		=guidance_scale > 1.0
				# get unconditional embeddings for classifier free guidance
				if do_classifier_free_guidance:
						lowerCamelCase :			Union[str, Any]    		=text_input.input_ids.shape[-1]
						lowerCamelCase :			str    		=self.tokenizer(['''''']      ,  padding='''max_length'''      ,  max_length=__lowercase      ,  return_tensors='''pt''' )
						lowerCamelCase :			Any    		=self.text_encoder(uncond_input.input_ids.to(self.device ) )[0]
						# duplicate unconditional embeddings for each generation per prompt
						lowerCamelCase :			str    		=uncond_embeddings.repeat_interleave(__lowercase      ,  dim=0 )

						# 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
						lowerCamelCase :			Optional[Any]    		=torch.cat([uncond_embeddings, text_embeddings] )

				# get the initial random noise unless the user supplied it

				# Unlike in other pipelines, latents need to be generated in the target device
				# for 1-to-1 results reproducibility with the CompVis implementation.
				# However this currently doesn't work in `mps`.
				lowerCamelCase :			Optional[int]    		=(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
				lowerCamelCase :			List[Any]    		=text_embeddings.dtype
				if latents is None:
						if self.device.type == "mps":
								# randn does not work reproducibly on mps
								lowerCamelCase :			Optional[int]    		=torch.randn(__lowercase      ,  generator=__lowercase      ,  device='''cpu'''      ,  dtype=__lowercase ).to(
								    self.device )
						else:
								lowerCamelCase :			int    		=torch.randn(__lowercase      ,  generator=__lowercase      ,  device=self.device      ,  dtype=__lowercase )
				else:
						if latents.shape != latents_shape:
								raise ValueError(F"Unexpected latents shape, got {latents.shape}, expected {latents_shape}" )
						lowerCamelCase :			Optional[int]    		=latents.to(self.device )

				# set timesteps
				lowerCamelCase :			Optional[Any]    		='''offset''' in set(inspect.signature(self.scheduler.set_timesteps ).parameters.keys() )
				lowerCamelCase :			List[Any]    		={}
				if accepts_offset:
						lowerCamelCase :			Tuple    		=1

				self.scheduler.set_timesteps(__lowercase      ,  **__lowercase )

				# Some schedulers like PNDM have timesteps as arrays
				# It's more optimized to move all timesteps to correct device beforehand
				lowerCamelCase :			Optional[int]    		=self.scheduler.timesteps.to(self.device )

				# scale the initial noise by the standard deviation required by the scheduler
				lowerCamelCase :			List[Any]    		=latents * self.scheduler.init_noise_sigma

				# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
				# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
				# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
				# and should be between [0, 1]
				lowerCamelCase :			Tuple    		='''eta''' in set(inspect.signature(self.scheduler.step ).parameters.keys() )
				lowerCamelCase :			Union[str, Any]    		={}
				if accepts_eta:
						lowerCamelCase :			List[str]    		=eta

				# check if the scheduler accepts generator
				lowerCamelCase :			Tuple    		='''generator''' in set(inspect.signature(self.scheduler.step ).parameters.keys() )
				if accepts_generator:
						lowerCamelCase :			int    		=generator

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

						# predict the noise residual
						lowerCamelCase :			Optional[int]    		=self.unet(__lowercase      ,  __lowercase      ,  encoder_hidden_states=__lowercase ).sample

						# perform classifier free guidance
						if do_classifier_free_guidance:
								lowerCamelCase		, lowerCamelCase :			Optional[int]    		=noise_pred.chunk(2 )
								lowerCamelCase :			Union[str, Any]    		=noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

						# perform clip guidance
						if clip_guidance_scale > 0:
								lowerCamelCase :			List[Any]    		=(
								    text_embeddings.chunk(2 )[1] if do_classifier_free_guidance else text_embeddings
								)
								lowerCamelCase		, lowerCamelCase :			Union[str, Any]    		=self.cond_fn(
								    __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  __lowercase      ,  )

						# compute the previous noisy sample x_t -> x_t-1
						lowerCamelCase :			Dict    		=self.scheduler.step(__lowercase      ,  __lowercase      ,  __lowercase      ,  **__lowercase ).prev_sample

				# scale and decode the image latents with vae
				lowerCamelCase :			Tuple    		=1 / self.vae.config.scaling_factor * latents
				lowerCamelCase :			str    		=self.vae.decode(__lowercase ).sample

				lowerCamelCase :			Optional[int]    		=(image / 2 + 0.5).clamp(0      ,  1 )
				lowerCamelCase :			Union[str, Any]    		=image.cpu().permute(0      ,  2      ,  3      ,  1 ).numpy()

				if output_type == "pil":
						lowerCamelCase :			Optional[Any]    		=self.numpy_to_pil(__lowercase )

				if not return_dict:
						return (image, None)

				return StableDiffusionPipelineOutput(images=__lowercase      ,  nsfw_content_detected=__lowercase )

