





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

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


A    :					Union[str, Any] =      logging.get_logger(__name__)  # pylint: disable=invalid-name


class   A			(SCREAMING_SNAKE_CASE    ):
			'''simple docstring'''




			def __init__(			self   :    str							,	__lowerCAmelCase   :    AutoencoderKL							,	__lowerCAmelCase   :    CLIPTextModel							,	__lowerCAmelCase   :    CLIPTokenizer							,	__lowerCAmelCase   :    UNetaDConditionModel							,	__lowerCAmelCase   :    Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]							,	__lowerCAmelCase   :    StableDiffusionSafetyChecker							,	__lowerCAmelCase   :    CLIPImageProcessor							,	)	->	int:







					"""simple docstring"""






					super().__init__()
					self.register_modules(
					    vae=__lowerCAmelCase							,	text_encoder=__lowerCAmelCase							,	tokenizer=__lowerCAmelCase							,	unet=__lowerCAmelCase							,	scheduler=__lowerCAmelCase							,	safety_checker=__lowerCAmelCase							,	feature_extractor=__lowerCAmelCase							,	)






			def    a_       (			self   :    List[Any]							,	__lowerCAmelCase   :    Optional[Union[str, int]] = "auto"   )	->	Dict:







					"""simple docstring"""






					if slice_size == "auto":
							# half the attention head size is usually a good trade-off between
							# speed and memory
							A__        =   self.unet.config.attention_head_dim // 2
					self.unet.set_attention_slice(__lowerCAmelCase   )






			def    a_       (			self   :    Optional[int]   )	->	str:







					"""simple docstring"""






					self.enable_attention_slicing(__lowerCAmelCase   )







			@torch.no_grad()
			def __call__(			self   :    List[Any]							,	__lowerCAmelCase   :    Union[str, List[str]]							,	__lowerCAmelCase   :    int = 5_12							,	__lowerCAmelCase   :    int = 5_12							,	__lowerCAmelCase   :    int = 50							,	__lowerCAmelCase   :    float = 7.5							,	__lowerCAmelCase   :    Optional[Union[str, List[str]]] = None							,	__lowerCAmelCase   :    Optional[int] = 1							,	__lowerCAmelCase   :    float = 0.0							,	__lowerCAmelCase   :    Optional[torch.Generator] = None							,	__lowerCAmelCase   :    Optional[torch.FloatTensor] = None							,	__lowerCAmelCase   :    Optional[str] = "pil"							,	__lowerCAmelCase   :    bool = True							,	__lowerCAmelCase   :    Optional[Callable[[int, int, torch.FloatTensor], None]] = None							,	__lowerCAmelCase   :    int = 1							,	__lowerCAmelCase   :    Optional[torch.FloatTensor] = None							,	**__lowerCAmelCase   :    Tuple							,	)	->	Optional[Any]:







					"""simple docstring"""






					if isinstance(__lowerCAmelCase							,	__lowerCAmelCase   ):
							A__        =   1
					elif isinstance(__lowerCAmelCase							,	__lowerCAmelCase   ):
							A__        =   len(__lowerCAmelCase   )
					else:
							raise ValueError(f'`prompt` has to be of type `str` or `list` but is {type(__lowerCAmelCase   )}'   )

					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}.'   )

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

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

					if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
							A__        =   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__        =   text_input_ids[:, : self.tokenizer.model_max_length]

					if text_embeddings is None:
							A__        =   self.text_encoder(text_input_ids.to(self.device   )   )[0]

					# duplicate text embeddings for each generation per prompt, using mps friendly method
					A__      ,			A__      ,			A__        =   text_embeddings.shape
					A__        =   text_embeddings.repeat(1							,	__lowerCAmelCase							,	1   )
					A__        =   text_embeddings.view(bs_embed * num_images_per_prompt							,	__lowerCAmelCase							,	-1   )

					# 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.
					A__        =   guidance_scale > 1.0
					# get unconditional embeddings for classifier free guidance
					if do_classifier_free_guidance:
							A__        =   42
							if negative_prompt is None:
									A__        =   [""""""]
							elif type(__lowerCAmelCase   ) is not type(__lowerCAmelCase   ):
									raise TypeError(
									    f'`negative_prompt` should be the same type to `prompt`, but got {type(__lowerCAmelCase   )} !='
									    f' {type(__lowerCAmelCase   )}.'   )
							elif isinstance(__lowerCAmelCase							,	__lowerCAmelCase   ):
									A__        =   [negative_prompt]
							elif batch_size != len(__lowerCAmelCase   ):
									raise ValueError(
									    f'`negative_prompt`: {negative_prompt} has batch size {len(__lowerCAmelCase   )}, but `prompt`:'
									    f' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
									    """ the batch size of `prompt`."""   )
							else:
									A__        =   negative_prompt

							A__        =   text_input_ids.shape[-1]
							A__        =   self.tokenizer(
							    __lowerCAmelCase							,	padding="""max_length"""							,	max_length=__lowerCAmelCase							,	truncation=__lowerCAmelCase							,	return_tensors="""pt"""							,	)
							A__        =   self.text_encoder(uncond_input.input_ids.to(self.device   )   )[0]

							# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
							A__        =   uncond_embeddings.shape[1]
							A__        =   uncond_embeddings.repeat(__lowerCAmelCase							,	__lowerCAmelCase							,	1   )
							A__        =   uncond_embeddings.view(batch_size * num_images_per_prompt							,	__lowerCAmelCase							,	-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__        =   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`.
					A__        =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
					A__        =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
					A__        =   text_embeddings.dtype
					if latents is None:
							if self.device.type == "mps":
									# randn does not exist on mps
									A__        =   torch.randn(
									    __lowerCAmelCase							,	generator=__lowerCAmelCase							,	device="""cpu"""							,	dtype=__lowerCAmelCase   ).to(self.device   )
									A__        =   torch.randn(__lowerCAmelCase							,	generator=__lowerCAmelCase							,	device="""cpu"""							,	dtype=__lowerCAmelCase   ).to(
									    self.device   )
							else:
									A__        =   torch.randn(
									    __lowerCAmelCase							,	generator=__lowerCAmelCase							,	device=self.device							,	dtype=__lowerCAmelCase   )
									A__        =   torch.randn(__lowerCAmelCase							,	generator=__lowerCAmelCase							,	device=self.device							,	dtype=__lowerCAmelCase   )
					else:
							if latents_reference.shape != latents_shape:
									raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'   )
							A__        =   latents_reference.to(self.device   )
							A__        =   latents.to(self.device   )

					# This is the key part of the pipeline where we
					# try to ensure that the generated images w/ the same seed
					# but different sizes actually result in similar images
					A__        =   (latents_shape[3] - latents_shape_reference[3]) // 2
					A__        =   (latents_shape[2] - latents_shape_reference[2]) // 2
					A__        =   latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
					A__        =   latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
					A__        =   0 if dx < 0 else dx
					A__        =   0 if dy < 0 else dy
					A__        =   max(-dx							,	0   )
					A__        =   max(-dy							,	0   )
					# import pdb
					# pdb.set_trace()
					A__        =   latents_reference[:, :, dy : dy + h, dx : dx + w]

					# set timesteps
					self.scheduler.set_timesteps(__lowerCAmelCase   )

					# Some schedulers like PNDM have timesteps as arrays
					# It's more optimized to move all timesteps to correct device beforehand
					A__        =   self.scheduler.timesteps.to(self.device   )

					# scale the initial noise by the standard deviation required by the scheduler
					A__        =   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]
					A__        =   """eta""" in set(inspect.signature(self.scheduler.step   ).parameters.keys()   )
					A__        =   {}
					if accepts_eta:
							A__        =   eta

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

							# predict the noise residual
							A__        =   self.unet(__lowerCAmelCase							,	__lowerCAmelCase							,	encoder_hidden_states=__lowerCAmelCase   ).sample

							# perform guidance
							if do_classifier_free_guidance:
									A__      ,			A__        =   noise_pred.chunk(2   )
									A__        =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

							# compute the previous noisy sample x_t -> x_t-1
							A__        =   self.scheduler.step(__lowerCAmelCase							,	__lowerCAmelCase							,	__lowerCAmelCase							,	**__lowerCAmelCase   ).prev_sample

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

					A__        =   1 / 0.1_8_2_1_5 * latents
					A__        =   self.vae.decode(__lowerCAmelCase   ).sample

					A__        =   (image / 2 + 0.5).clamp(0							,	1   )

					# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
					A__        =   image.cpu().permute(0							,	2							,	3							,	1   ).float().numpy()

					if self.safety_checker is not None:
							A__        =   self.feature_extractor(self.numpy_to_pil(__lowerCAmelCase   )							,	return_tensors="""pt"""   ).to(
							    self.device   )
							A__      ,			A__        =   self.safety_checker(
							    images=__lowerCAmelCase							,	clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype   )   )
					else:
							A__        =   None

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

					if not return_dict:
							return (image, has_nsfw_concept)

					return StableDiffusionPipelineOutput(images=__lowerCAmelCase							,	nsfw_content_detected=__lowerCAmelCase   )










import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import CLIPTokenizer, CLIPTokenizerFast
from transformers.models.clip.tokenization_clip import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
					from PIL import Image

					from transformers import CLIPImageProcessor, CLIPProcessor


@require_vision
class   A			(unittest.TestCase    ):
			'''simple docstring'''




			def    a_       (			self   :    Union[str, Any]   )	->	Dict:







					"""simple docstring"""






					A__        =   tempfile.mkdtemp()

					# fmt: off
					A__        =   ["""l""", """o""", """w""", """e""", """r""", """s""", """t""", """i""", """d""", """n""", """lo""", """l</w>""", """w</w>""", """r</w>""", """t</w>""", """low</w>""", """er</w>""", """lowest</w>""", """newer</w>""", """wider""", """<unk>""", """<|startoftext|>""", """<|endoftext|>"""]
					# fmt: on
					A__        =   dict(zip(__lowerCAmelCase							,	range(len(__lowerCAmelCase   )   )   )   )
					A__        =   ["""#version: 0.2""", """l o""", """lo w</w>""", """e r</w>""", """"""]
					A__        =   {"""unk_token""": """<unk>"""}

					A__        =   os.path.join(self.tmpdirname							,	VOCAB_FILES_NAMES["""vocab_file"""]   )
					A__        =   os.path.join(self.tmpdirname							,	VOCAB_FILES_NAMES["""merges_file"""]   )
					with open(self.vocab_file							,	"""w"""							,	encoding="""utf-8"""   ) as fp:
							fp.write(json.dumps(__lowerCAmelCase   ) + """\n"""   )
					with open(self.merges_file							,	"""w"""							,	encoding="""utf-8"""   ) as fp:
							fp.write("""\n""".join(__lowerCAmelCase   )   )

					A__        =   {
					    """do_resize""": True,
					    """size""": 20,
					    """do_center_crop""": True,
					    """crop_size""": 18,
					    """do_normalize""": True,
					    """image_mean""": [0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3],
					    """image_std""": [0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1],
					}
					A__        =   os.path.join(self.tmpdirname							,	__lowerCAmelCase   )
					with open(self.image_processor_file							,	"""w"""							,	encoding="""utf-8"""   ) as fp:
							json.dump(__lowerCAmelCase							,	__lowerCAmelCase   )






			def    a_       (			self   :    Tuple							,	**__lowerCAmelCase   :    Dict   )	->	str:







					"""simple docstring"""






					return CLIPTokenizer.from_pretrained(self.tmpdirname							,	**__lowerCAmelCase   )






			def    a_       (			self   :    Union[str, Any]							,	**__lowerCAmelCase   :    Dict   )	->	List[str]:







					"""simple docstring"""






					return CLIPTokenizerFast.from_pretrained(self.tmpdirname							,	**__lowerCAmelCase   )






			def    a_       (			self   :    List[str]							,	**__lowerCAmelCase   :    Optional[Any]   )	->	Dict:







					"""simple docstring"""






					return CLIPImageProcessor.from_pretrained(self.tmpdirname							,	**__lowerCAmelCase   )






			def    a_       (			self   :    str   )	->	Dict:







					"""simple docstring"""






					shutil.rmtree(self.tmpdirname   )






			def    a_       (			self   :    str   )	->	Any:







					"""simple docstring"""






					A__        =   [np.random.randint(2_55							,	size=(3, 30, 4_00)							,	dtype=np.uinta   )]

					A__        =   [Image.fromarray(np.moveaxis(__lowerCAmelCase							,	0							,	-1   )   ) for x in image_inputs]

					return image_inputs






			def    a_       (			self   :    Optional[int]   )	->	Tuple:







					"""simple docstring"""






					A__        =   self.get_tokenizer()
					A__        =   self.get_rust_tokenizer()
					A__        =   self.get_image_processor()

					A__        =   CLIPProcessor(tokenizer=__lowerCAmelCase							,	image_processor=__lowerCAmelCase   )
					processor_slow.save_pretrained(self.tmpdirname   )
					A__        =   CLIPProcessor.from_pretrained(self.tmpdirname							,	use_fast=__lowerCAmelCase   )

					A__        =   CLIPProcessor(tokenizer=__lowerCAmelCase							,	image_processor=__lowerCAmelCase   )
					processor_fast.save_pretrained(self.tmpdirname   )
					A__        =   CLIPProcessor.from_pretrained(self.tmpdirname   )

					self.assertEqual(processor_slow.tokenizer.get_vocab()							,	tokenizer_slow.get_vocab()   )
					self.assertEqual(processor_fast.tokenizer.get_vocab()							,	tokenizer_fast.get_vocab()   )
					self.assertEqual(tokenizer_slow.get_vocab()							,	tokenizer_fast.get_vocab()   )
					self.assertIsInstance(processor_slow.tokenizer							,	__lowerCAmelCase   )
					self.assertIsInstance(processor_fast.tokenizer							,	__lowerCAmelCase   )

					self.assertEqual(processor_slow.image_processor.to_json_string()							,	image_processor.to_json_string()   )
					self.assertEqual(processor_fast.image_processor.to_json_string()							,	image_processor.to_json_string()   )
					self.assertIsInstance(processor_slow.image_processor							,	__lowerCAmelCase   )
					self.assertIsInstance(processor_fast.image_processor							,	__lowerCAmelCase   )






			def    a_       (			self   :    Optional[Any]   )	->	Union[str, Any]:







					"""simple docstring"""






					A__        =   CLIPProcessor(tokenizer=self.get_tokenizer()							,	image_processor=self.get_image_processor()   )
					processor.save_pretrained(self.tmpdirname   )

					A__        =   self.get_tokenizer(bos_token="""(BOS)"""							,	eos_token="""(EOS)"""   )
					A__        =   self.get_image_processor(do_normalize=__lowerCAmelCase							,	padding_value=1.0   )

					A__        =   CLIPProcessor.from_pretrained(
					    self.tmpdirname							,	bos_token="""(BOS)"""							,	eos_token="""(EOS)"""							,	do_normalize=__lowerCAmelCase							,	padding_value=1.0   )

					self.assertEqual(processor.tokenizer.get_vocab()							,	tokenizer_add_kwargs.get_vocab()   )
					self.assertIsInstance(processor.tokenizer							,	__lowerCAmelCase   )

					self.assertEqual(processor.image_processor.to_json_string()							,	image_processor_add_kwargs.to_json_string()   )
					self.assertIsInstance(processor.image_processor							,	__lowerCAmelCase   )






			def    a_       (			self   :    List[Any]   )	->	Dict:







					"""simple docstring"""






					A__        =   self.get_image_processor()
					A__        =   self.get_tokenizer()

					A__        =   CLIPProcessor(tokenizer=__lowerCAmelCase							,	image_processor=__lowerCAmelCase   )

					A__        =   self.prepare_image_inputs()

					A__        =   image_processor(__lowerCAmelCase							,	return_tensors="""np"""   )
					A__        =   processor(images=__lowerCAmelCase							,	return_tensors="""np"""   )

					for key in input_image_proc.keys():
							self.assertAlmostEqual(input_image_proc[key].sum()							,	input_processor[key].sum()							,	delta=1e-2   )






			def    a_       (			self   :    Optional[Any]   )	->	Any:







					"""simple docstring"""






					A__        =   self.get_image_processor()
					A__        =   self.get_tokenizer()

					A__        =   CLIPProcessor(tokenizer=__lowerCAmelCase							,	image_processor=__lowerCAmelCase   )

					A__        =   """lower newer"""

					A__        =   processor(text=__lowerCAmelCase   )

					A__        =   tokenizer(__lowerCAmelCase   )

					for key in encoded_tok.keys():
							self.assertListEqual(encoded_tok[key]							,	encoded_processor[key]   )






			def    a_       (			self   :    Union[str, Any]   )	->	Dict:







					"""simple docstring"""






					A__        =   self.get_image_processor()
					A__        =   self.get_tokenizer()

					A__        =   CLIPProcessor(tokenizer=__lowerCAmelCase							,	image_processor=__lowerCAmelCase   )

					A__        =   """lower newer"""
					A__        =   self.prepare_image_inputs()

					A__        =   processor(text=__lowerCAmelCase							,	images=__lowerCAmelCase   )

					self.assertListEqual(list(inputs.keys()   )							,	["""input_ids""", """attention_mask""", """pixel_values"""]   )

					# test if it raises when no input is passed
					with pytest.raises(__lowerCAmelCase   ):
							processor()






			def    a_       (			self   :    Tuple   )	->	str:







					"""simple docstring"""






					A__        =   self.get_image_processor()
					A__        =   self.get_tokenizer()

					A__        =   CLIPProcessor(tokenizer=__lowerCAmelCase							,	image_processor=__lowerCAmelCase   )

					A__        =   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

					A__        =   processor.batch_decode(__lowerCAmelCase   )
					A__        =   tokenizer.batch_decode(__lowerCAmelCase   )

					self.assertListEqual(__lowerCAmelCase							,	__lowerCAmelCase   )







			def    a_       (			self   :    Optional[int]   )	->	str:







					"""simple docstring"""






					A__        =   self.get_image_processor()
					A__        =   self.get_tokenizer()

					A__        =   CLIPProcessor(tokenizer=__lowerCAmelCase							,	image_processor=__lowerCAmelCase   )

					A__        =   """lower newer"""
					A__        =   self.prepare_image_inputs()

					A__        =   processor(text=__lowerCAmelCase							,	images=__lowerCAmelCase   )

					self.assertListEqual(list(inputs.keys()   )							,	processor.model_input_names   )



