

import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaImgaImgPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()



class   __lowercase    (			UpperCAmelCase_		,	unittest.TestCase     ):




			"""simple docstring"""







			_UpperCAmelCase       :							str      							=					KandinskyVaaImgaImgPipeline
			_UpperCAmelCase       :							Dict      							=					['''image_embeds''', '''negative_image_embeds''', '''image''']
			_UpperCAmelCase       :							List[str]      							=					[
			    '''image_embeds''',
			    '''negative_image_embeds''',
			    '''image''',
			]
			_UpperCAmelCase       :							Optional[int]      							=					[
			    '''generator''',
			    '''height''',
			    '''width''',
			    '''strength''',
			    '''guidance_scale''',
			    '''num_inference_steps''',
			    '''return_dict''',
			    '''guidance_scale''',
			    '''num_images_per_prompt''',
			    '''output_type''',
			    '''return_dict''',
			]
			_UpperCAmelCase       :							Optional[Any]      							=					False



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					Any):
									return 32



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[Any]):
									return 32



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					Dict):
									return self.time_input_dim



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					Union[str, Any]):
									return self.time_input_dim * 4



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					str):
									return 100



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[Any]):
									torch.manual_seed(0)

									SCREAMING_SNAKE_CASE_:			Any  		=	{
									    "in_channels": 4,
									    # Out channels is double in channels because predicts mean and variance
									    "out_channels": 8,
									    "addition_embed_type": "image",
									    "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
									    "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
									    "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
									    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
									    "layers_per_block": 1,
									    "encoder_hid_dim": self.text_embedder_hidden_size,
									    "encoder_hid_dim_type": "image_proj",
									    "cross_attention_dim": self.cross_attention_dim,
									    "attention_head_dim": 4,
									    "resnet_time_scale_shift": "scale_shift",
									    "class_embed_type": None,
									}

									SCREAMING_SNAKE_CASE_:			Tuple  		=	UNetaDConditionModel(**lowerCAmelCase__)
									return model



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					Any):
									return {
									    "block_out_channels": [32, 64],
									    "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
									    "in_channels": 3,
									    "latent_channels": 4,
									    "layers_per_block": 1,
									    "norm_num_groups": 8,
									    "norm_type": "spatial",
									    "num_vq_embeddings": 12,
									    "out_channels": 3,
									    "up_block_types": [
									        "AttnUpDecoderBlock2D",
									        "UpDecoderBlock2D",
									    ],
									    "vq_embed_dim": 4,
									}



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[Any]):
									torch.manual_seed(0)
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	VQModel(**self.dummy_movq_kwargs)
									return model



			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[str]):
									SCREAMING_SNAKE_CASE_:			str  		=	self.dummy_unet
									SCREAMING_SNAKE_CASE_:			str  		=	self.dummy_movq

									SCREAMING_SNAKE_CASE_:			Optional[int]  		=	{
									    "num_train_timesteps": 1000,
									    "beta_schedule": "linear",
									    "beta_start": 0.0_0085,
									    "beta_end": 0.012,
									    "clip_sample": False,
									    "set_alpha_to_one": False,
									    "steps_offset": 0,
									    "prediction_type": "epsilon",
									    "thresholding": False,
									}

									SCREAMING_SNAKE_CASE_:			Tuple  		=	DDIMScheduler(**lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_:			Optional[int]  		=	{
									    "unet": unet,
									    "scheduler": scheduler,
									    "movq": movq,
									}

									return components



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Tuple       ,						lowerCAmelCase__  :					str       ,						lowerCAmelCase__  :					str=0):
									SCREAMING_SNAKE_CASE_:			Optional[int]  		=	floats_tensor((1, self.text_embedder_hidden_size)       ,						rng=random.Random(lowerCAmelCase__)).to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	floats_tensor((1, self.text_embedder_hidden_size)       ,						rng=random.Random(seed + 1)).to(
									    lowerCAmelCase__)
									# create init_image
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	floats_tensor((1, 3, 64, 64)       ,						rng=random.Random(lowerCAmelCase__)).to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	image.cpu().permute(0       ,						2       ,						3       ,						1)[0]
									SCREAMING_SNAKE_CASE_:			Dict  		=	Image.fromarray(np.uinta(lowerCAmelCase__)).convert("RGB").resize((256, 256))

									if str(lowerCAmelCase__).startswith("mps"):
															SCREAMING_SNAKE_CASE_:			str  		=	torch.manual_seed(lowerCAmelCase__)
									else:
															SCREAMING_SNAKE_CASE_:			Any  		=	torch.Generator(device=lowerCAmelCase__).manual_seed(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Tuple  		=	{
									    "image": init_image,
									    "image_embeds": image_embeds,
									    "negative_image_embeds": negative_image_embeds,
									    "generator": generator,
									    "height": 64,
									    "width": 64,
									    "num_inference_steps": 10,
									    "guidance_scale": 7.0,
									    "strength": 0.2,
									    "output_type": "np",
									}
									return inputs




			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[int]):
									SCREAMING_SNAKE_CASE_:			Any  		=	"cpu"

									SCREAMING_SNAKE_CASE_:			Any  		=	self.get_dummy_components()

									SCREAMING_SNAKE_CASE_:			List[str]  		=	self.pipeline_class(**lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Dict  		=	pipe.to(lowerCAmelCase__)

									pipe.set_progress_bar_config(disable=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_:			List[Any]  		=	pipe(**self.get_dummy_inputs(lowerCAmelCase__))
									SCREAMING_SNAKE_CASE_:			Dict  		=	output.images

									SCREAMING_SNAKE_CASE_:			List[str]  		=	pipe(
									    **self.get_dummy_inputs(lowerCAmelCase__)       ,						return_dict=lowerCAmelCase__       ,						)[0]

									SCREAMING_SNAKE_CASE_:			List[Any]  		=	image[0, -3:, -3:, -1]
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	image_from_tuple[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)

									SCREAMING_SNAKE_CASE_:			Tuple  		=	np.array(
									    [0.619_9778, 0.6398_4406, 0.4614_5785, 0.6294_4984, 0.562_2215, 0.4730_6132, 0.4744_1456, 0.460_7606, 0.4871_9263])
									assert (
									    np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
									), F" expected_slice {expected_slice}, but got {image_slice.flatten()}"
									assert (
									    np.abs(image_from_tuple_slice.flatten() - expected_slice).max() < 1E-2
									), F" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"







@slow
@require_torch_gpu
class   __lowercase    (			unittest.TestCase     ):




			"""simple docstring"""







			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[str]):
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()




			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[Any]):
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	load_numpy(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
									    "/kandinskyv22/kandinskyv22_img2img_frog.npy")

									SCREAMING_SNAKE_CASE_:			List[str]  		=	load_image(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png")
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	"A red cartoon frog, 4k"

									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	KandinskyVaaPriorPipeline.from_pretrained(
									    "kandinsky-community/kandinsky-2-2-prior"       ,						torch_dtype=torch.floataa)
									pipe_prior.to(lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	KandinskyVaaImgaImgPipeline.from_pretrained(
									    "kandinsky-community/kandinsky-2-2-decoder"       ,						torch_dtype=torch.floataa)
									SCREAMING_SNAKE_CASE_:			Tuple  		=	pipeline.to(lowerCAmelCase__)

									pipeline.set_progress_bar_config(disable=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_:			int  		=	torch.Generator(device="cpu").manual_seed(0)
									SCREAMING_SNAKE_CASE_		,   SCREAMING_SNAKE_CASE_:			List[Any]  		=	pipe_prior(
									    lowerCAmelCase__       ,						generator=lowerCAmelCase__       ,						num_inference_steps=5       ,						negative_prompt=""       ,						).to_tuple()

									SCREAMING_SNAKE_CASE_:			str  		=	pipeline(
									    image=lowerCAmelCase__       ,						image_embeds=lowerCAmelCase__       ,						negative_image_embeds=lowerCAmelCase__       ,						generator=lowerCAmelCase__       ,						num_inference_steps=100       ,						height=768       ,						width=768       ,						strength=0.2       ,						output_type="np"       ,						)

									SCREAMING_SNAKE_CASE_:			Tuple  		=	output.images[0]

									assert image.shape == (768, 768, 3)

									assert_mean_pixel_difference(lowerCAmelCase__       ,						lowerCAmelCase__)









import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
						import torch

						from transformers import AutoModelForCausalLM



@require_torch
class   __lowercase    (			unittest.TestCase     ):




			"""simple docstring"""







			def 	_SCREAMING_SNAKE_CASE      (			self  :					int):
									SCREAMING_SNAKE_CASE_:			str  		=	AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2")
									SCREAMING_SNAKE_CASE_:			str  		=	AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2").to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			str  		=	-1

									SCREAMING_SNAKE_CASE_:			Optional[int]  		=	ids_tensor((1, 5)       ,						vocab_size=model.config.vocab_size).to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Any  		=	model.generate(lowerCAmelCase__       ,						max_new_tokens=10       ,						do_sample=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			List[str]  		=	tokenizer.decode(greedy_ids[0])

									with CaptureStdout() as cs:
															SCREAMING_SNAKE_CASE_:			int  		=	TextStreamer(lowerCAmelCase__)
															model.generate(lowerCAmelCase__       ,						max_new_tokens=10       ,						do_sample=lowerCAmelCase__       ,						streamer=lowerCAmelCase__)
									# The greedy text should be printed to stdout, except for the final "\n" in the streamer
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	cs.out[:-1]

									self.assertEqual(lowerCAmelCase__       ,						lowerCAmelCase__)



			def 	_SCREAMING_SNAKE_CASE      (			self  :					int):
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2")
									SCREAMING_SNAKE_CASE_:			int  		=	AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2").to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	-1

									SCREAMING_SNAKE_CASE_:			int  		=	ids_tensor((1, 5)       ,						vocab_size=model.config.vocab_size).to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			int  		=	model.generate(lowerCAmelCase__       ,						max_new_tokens=10       ,						do_sample=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Tuple  		=	tokenizer.decode(greedy_ids[0])

									SCREAMING_SNAKE_CASE_:			int  		=	TextIteratorStreamer(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	{"input_ids": input_ids, "max_new_tokens": 10, "do_sample": False, "streamer": streamer}
									SCREAMING_SNAKE_CASE_:			Tuple  		=	Thread(target=model.generate       ,						kwargs=lowerCAmelCase__)
									thread.start()
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	""
									for new_text in streamer:
															streamer_text += new_text

									self.assertEqual(lowerCAmelCase__       ,						lowerCAmelCase__)



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[int]):
									SCREAMING_SNAKE_CASE_:			int  		=	AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2")
									SCREAMING_SNAKE_CASE_:			int  		=	AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2").to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	-1

									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	ids_tensor((1, 5)       ,						vocab_size=model.config.vocab_size).to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Dict  		=	model.generate(lowerCAmelCase__       ,						max_new_tokens=10       ,						do_sample=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Tuple  		=	greedy_ids[:, input_ids.shape[1] :]
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	tokenizer.decode(new_greedy_ids[0])

									with CaptureStdout() as cs:
															SCREAMING_SNAKE_CASE_:			Dict  		=	TextStreamer(lowerCAmelCase__       ,						skip_prompt=lowerCAmelCase__)
															model.generate(lowerCAmelCase__       ,						max_new_tokens=10       ,						do_sample=lowerCAmelCase__       ,						streamer=lowerCAmelCase__)
									# The greedy text should be printed to stdout, except for the final "\n" in the streamer
									SCREAMING_SNAKE_CASE_:			Any  		=	cs.out[:-1]

									self.assertEqual(lowerCAmelCase__       ,						lowerCAmelCase__)



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Tuple):
									# Tests that we can pass `decode_kwargs` to the streamer to control how the tokens are decoded. Must be tested
									# with actual models -- the dummy models' tokenizers are not aligned with their models, and
									# `skip_special_tokens=True` has no effect on them
									SCREAMING_SNAKE_CASE_:			Tuple  		=	AutoTokenizer.from_pretrained("distilgpt2")
									SCREAMING_SNAKE_CASE_:			List[str]  		=	AutoModelForCausalLM.from_pretrained("distilgpt2").to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	-1

									SCREAMING_SNAKE_CASE_:			List[str]  		=	torch.ones((1, 5)       ,						device=lowerCAmelCase__).long() * model.config.bos_token_id
									with CaptureStdout() as cs:
															SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	TextStreamer(lowerCAmelCase__       ,						skip_special_tokens=lowerCAmelCase__)
															model.generate(lowerCAmelCase__       ,						max_new_tokens=1       ,						do_sample=lowerCAmelCase__       ,						streamer=lowerCAmelCase__)

									# The prompt contains a special token, so the streamer should not print it. As such, the output text, when
									# re-tokenized, must only contain one token
									SCREAMING_SNAKE_CASE_:			str  		=	cs.out[:-1]  # Remove the final "\n"
									SCREAMING_SNAKE_CASE_:			Tuple  		=	tokenizer(lowerCAmelCase__       ,						return_tensors="pt")
									self.assertEqual(streamer_text_tokenized.input_ids.shape       ,						(1, 1))




			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[Any]):
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2")
									SCREAMING_SNAKE_CASE_:			List[str]  		=	AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2").to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Dict  		=	-1

									SCREAMING_SNAKE_CASE_:			List[str]  		=	ids_tensor((1, 5)       ,						vocab_size=model.config.vocab_size).to(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Any  		=	TextIteratorStreamer(lowerCAmelCase__       ,						timeout=0.001)
									SCREAMING_SNAKE_CASE_:			Any  		=	{"input_ids": input_ids, "max_new_tokens": 10, "do_sample": False, "streamer": streamer}
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	Thread(target=model.generate       ,						kwargs=lowerCAmelCase__)
									thread.start()

									# The streamer will timeout after 0.001 seconds, so an exception will be raised
									with self.assertRaises(lowerCAmelCase__):
															SCREAMING_SNAKE_CASE_:			Tuple  		=	""
															for new_text in streamer:
																					streamer_text += new_text






