




"""simple docstring"""




import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetImgaImgPipeline,
    KandinskyVaaPriorEmbaEmbPipeline,
    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     __magic_name__							(       UpperCAmelCase__						,					unittest.TestCase				):






				'''simple docstring'''
				__UpperCamelCase			  =						KandinskyVaaControlnetImgaImgPipeline
				__UpperCamelCase			  =						["image_embeds", "negative_image_embeds", "image", "hint"]
				__UpperCamelCase			  =						["image_embeds", "negative_image_embeds", "image", "hint"]
				__UpperCamelCase			  =						[
				    "generator",
				    "height",
				    "width",
				    "strength",
				    "guidance_scale",
				    "num_inference_steps",
				    "return_dict",
				    "guidance_scale",
				    "num_images_per_prompt",
				    "output_type",
				    "return_dict",
				]
				__UpperCamelCase			  =						False





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							return 32





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							return 32





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							return self.time_input_dim





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							return self.time_input_dim * 4





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							return 100





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							torch.manual_seed(0 )

							lowerCamelCase			      =					{
							    """in_channels""": 8,
							    # Out channels is double in channels because predicts mean and variance
							    """out_channels""": 8,
							    """addition_embed_type""": """image_hint""",
							    """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,
							}

							lowerCamelCase			      =					UNetaDConditionModel(**_a )
							return model





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							return {
							    "block_out_channels": [32, 32, 64, 64],
							    "down_block_types": [
							        "DownEncoderBlock2D",
							        "DownEncoderBlock2D",
							        "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", "UpDecoderBlock2D", "UpDecoderBlock2D"],
							    "vq_embed_dim": 4,
							}





				@property
				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							torch.manual_seed(0 )
							lowerCamelCase			      =					VQModel(**self.dummy_movq_kwargs )
							return model





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					self.dummy_unet
							lowerCamelCase			      =					self.dummy_movq

							lowerCamelCase			      =					{
							    """num_train_timesteps""": 1_000,
							    """beta_schedule""": """linear""",
							    """beta_start""": 0.00_085,
							    """beta_end""": 0.012,
							    """clip_sample""": False,
							    """set_alpha_to_one""": False,
							    """steps_offset""": 0,
							    """prediction_type""": """epsilon""",
							    """thresholding""": False,
							}

							lowerCamelCase			      =					DDIMScheduler(**_a )

							lowerCamelCase			      =					{
							    """unet""": unet,
							    """scheduler""": scheduler,
							    """movq""": movq,
							}

							return components





				def    _lowerCAmelCase				(       self		, _a		, _a=0 ):





							"""simple docstring"""
							lowerCamelCase			      =					floats_tensor((1, self.text_embedder_hidden_size)		, rng=random.Random(_a ) ).to(_a )
							lowerCamelCase			      =					floats_tensor((1, self.text_embedder_hidden_size)		, rng=random.Random(seed + 1 ) ).to(
							    _a )
							# create init_image
							lowerCamelCase			      =					floats_tensor((1, 3, 64, 64)		, rng=random.Random(_a ) ).to(_a )
							lowerCamelCase			      =					image.cpu().permute(0		, 2		, 3		, 1 )[0]
							lowerCamelCase			      =					Image.fromarray(np.uinta(_a ) ).convert("""RGB""" ).resize((256, 256) )
							# create hint
							lowerCamelCase			      =					floats_tensor((1, 3, 64, 64)		, rng=random.Random(_a ) ).to(_a )

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





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					"""cpu"""

							lowerCamelCase			      =					self.get_dummy_components()

							lowerCamelCase			      =					self.pipeline_class(**_a )
							lowerCamelCase			      =					pipe.to(_a )

							pipe.set_progress_bar_config(disable=_a )

							lowerCamelCase			      =					pipe(**self.get_dummy_inputs(_a ) )
							lowerCamelCase			      =					output.images

							lowerCamelCase			      =					pipe(
							    **self.get_dummy_inputs(_a )		, return_dict=_a		, )[0]

							lowerCamelCase			      =					image[0, -3:, -3:, -1]
							lowerCamelCase			      =					image_from_tuple[0, -3:, -3:, -1]

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

							lowerCamelCase			      =					np.array(
							    [0.54_985_034, 0.55_509_365, 0.52_561_504, 0.5_570_494, 0.5_593_818, 0.5_263_979, 0.50_285_643, 0.5_069_846, 0.51_196_736] )
							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     __magic_name__							(       unittest.TestCase				):






				'''simple docstring'''





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							# clean up the VRAM after each test
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					load_numpy(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
							    """/kandinskyv22/kandinskyv22_controlnet_img2img_robotcat_fp16.npy""" )

							lowerCamelCase			      =					load_image(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/kandinsky/cat.png""" )
							lowerCamelCase			      =					init_image.resize((512, 512) )

							lowerCamelCase			      =					load_image(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
							    """/kandinskyv22/hint_image_cat.png""" )
							lowerCamelCase			      =					torch.from_numpy(np.array(_a ) ).float() / 255.0
							lowerCamelCase			      =					hint.permute(2		, 0		, 1 ).unsqueeze(0 )

							lowerCamelCase			      =					"""A robot, 4k photo"""

							lowerCamelCase			      =					KandinskyVaaPriorEmbaEmbPipeline.from_pretrained(
							    """kandinsky-community/kandinsky-2-2-prior"""		, torch_dtype=torch.floataa )
							pipe_prior.to(_a )

							lowerCamelCase			      =					KandinskyVaaControlnetImgaImgPipeline.from_pretrained(
							    """kandinsky-community/kandinsky-2-2-controlnet-depth"""		, torch_dtype=torch.floataa )
							lowerCamelCase			      =					pipeline.to(_a )

							pipeline.set_progress_bar_config(disable=_a )

							lowerCamelCase			      =					torch.Generator(device="""cpu""" ).manual_seed(0 )

							lowerCamelCase      ,   lowerCamelCase			      =					pipe_prior(
							    _a		, image=_a		, strength=0.85		, generator=_a		, negative_prompt=""""""		, ).to_tuple()

							lowerCamelCase			      =					pipeline(
							    image=_a		, image_embeds=_a		, negative_image_embeds=_a		, hint=_a		, generator=_a		, num_inference_steps=100		, height=512		, width=512		, strength=0.5		, output_type="""np"""		, )

							lowerCamelCase			      =					output.images[0]

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

							assert_mean_pixel_difference(_a		, _a )









"""simple docstring"""




import numpy as np
import torch
from torch.utils.data import Dataset, IterableDataset

from ..utils.generic import ModelOutput






class     __magic_name__							(       UpperCAmelCase__				):






				'''simple docstring'''





				def __init__(       self		, _a		, _a		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					dataset
							lowerCamelCase			      =					process
							lowerCamelCase			      =					params





				def __len__(       self ):





							"""simple docstring"""
							return len(self.dataset )





				def __getitem__(       self		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					self.dataset[i]
							lowerCamelCase			      =					self.process(_a		, **self.params )
							return processed






class     __magic_name__							(       UpperCAmelCase__				):






				'''simple docstring'''





				def __init__(       self		, _a		, _a		, _a		, _a=None ):





							"""simple docstring"""
							lowerCamelCase			      =					loader
							lowerCamelCase			      =					infer
							lowerCamelCase			      =					params
							if loader_batch_size == 1:
										# Let's spare some time by deactivating altogether
										lowerCamelCase			      =					None
							lowerCamelCase			      =					loader_batch_size

							# Internal bookkeeping
							lowerCamelCase			      =					None
							lowerCamelCase			      =					None





				def __len__(       self ):





							"""simple docstring"""
							return len(self.loader )





				def __iter__(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					iter(self.loader )
							return self





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							if isinstance(self._loader_batch_data		, torch.Tensor ):
										# Batch data is simple tensor, just fetch the slice
										lowerCamelCase			      =					self._loader_batch_data[self._loader_batch_index]
							else:
										# Batch data is assumed to be BaseModelOutput (or dict)
										lowerCamelCase			      =					{}
										for k, element in self._loader_batch_data.items():
													if isinstance(_a		, _a ):
																# Convert ModelOutput to tuple first
																lowerCamelCase			      =					element.to_tuple()
																if isinstance(element[0]		, torch.Tensor ):
																			lowerCamelCase			      =					tuple(el[self._loader_batch_index].unsqueeze(0 ) for el in element )
																elif isinstance(element[0]		, np.ndarray ):
																			lowerCamelCase			      =					tuple(np.expand_dims(el[self._loader_batch_index]		, 0 ) for el in element )
																continue
													if k in {"hidden_states", "past_key_values", "attentions"} and isinstance(_a		, _a ):
																# Those are stored as lists of tensors so need specific unbatching.
																if isinstance(element[0]		, torch.Tensor ):
																			lowerCamelCase			      =					tuple(el[self._loader_batch_index].unsqueeze(0 ) for el in element )
																elif isinstance(element[0]		, np.ndarray ):
																			lowerCamelCase			      =					tuple(np.expand_dims(el[self._loader_batch_index]		, 0 ) for el in element )
																continue
													if element is None:
																# This can happen for optional data that get passed around
																lowerCamelCase			      =					None
													elif isinstance(element[self._loader_batch_index]		, torch.Tensor ):
																# Take correct batch data, but make it looked like batch_size=1
																# For compatibility with other methods within transformers

																lowerCamelCase			      =					element[self._loader_batch_index].unsqueeze(0 )
													elif isinstance(element[self._loader_batch_index]		, np.ndarray ):
																# Take correct batch data, but make it looked like batch_size=1
																# For compatibility with other methods within transformers
																lowerCamelCase			      =					np.expand_dims(element[self._loader_batch_index]		, 0 )
													else:
																# This is typically a list, so no need to `unsqueeze`.
																lowerCamelCase			      =					element[self._loader_batch_index]
            # Recreate the element by reusing the original class to make it look
            # batch_size=1
										lowerCamelCase			      =					self._loader_batch_data.__class__(_a )
							self._loader_batch_index += 1
							return result





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							if self._loader_batch_index is not None and self._loader_batch_index < self.loader_batch_size:
										# We are currently unrolling a batch so we just need to return
										# the current item within a batch
										return self.loader_batch_item()

							# We're out of items within a batch
							lowerCamelCase			      =					next(self.iterator )
							lowerCamelCase			      =					self.infer(_a		, **self.params )
							# We now have a batch of "inferred things".
							if self.loader_batch_size is not None:
										# Try to infer the size of the batch
										if isinstance(_a		, torch.Tensor ):
													lowerCamelCase			      =					processed
										else:
													lowerCamelCase			      =					list(processed.keys() )[0]
													lowerCamelCase			      =					processed[key]
										if isinstance(_a		, _a ):
													lowerCamelCase			      =					len(_a )
										else:
													lowerCamelCase			      =					first_tensor.shape[0]
										if 0 < observed_batch_size < self.loader_batch_size:
													# could be last batch so we can't unroll as many
													# elements.
													lowerCamelCase			      =					observed_batch_size
										# Setting internal index to unwrap the batch
										lowerCamelCase			      =					processed
										lowerCamelCase			      =					0
										return self.loader_batch_item()
							else:
										# We're not unrolling batches
										return processed






class     __magic_name__							(       UpperCAmelCase__				):






				'''simple docstring'''





				def __init__(       self		, _a		, _a		, _a		, _a=None ):





							"""simple docstring"""
							super().__init__(_a		, _a		, _a )





				def __iter__(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					iter(self.loader )
							lowerCamelCase			      =					None
							return self





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							if self.subiterator is None:
										lowerCamelCase			      =					self.infer(next(self.iterator )		, **self.params )
							try:
										# Try to return next item
										lowerCamelCase			      =					next(self.subiterator )
							except StopIteration:
										# When a preprocess iterator ends, we can start lookig at the next item
										# ChunkIterator will keep feeding until ALL elements of iterator
										# all have created their subiterator and have been iterating against.
										#
										# Another way to look at it, is we're basically flattening lists of lists
										# into a single list, but with generators
										lowerCamelCase			      =					self.infer(next(self.iterator )		, **self.params )
										lowerCamelCase			      =					next(self.subiterator )
							return processed






class     __magic_name__							(       UpperCAmelCase__				):






				'''simple docstring'''





				def __iter__(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					iter(self.loader )
							return self





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							# Extremely similar to PipelineIterator in its unpacking mechanism
							# BUT, we have an extra required item which is the presence of `is_last`
							# That is because everything is flattened by `PipelineChunkIterator` we
							# need to keep track of how to regroup here in the original `process`
							# boundaries so that `process` and `postprocess` see the same data.

							# This iterator accumulates items (possibly while unbatching) until it
							# its a `is_last` and then just passes it on to the caller.
							lowerCamelCase			      =					False
							lowerCamelCase			      =					[]
							if self._loader_batch_index is not None and self._loader_batch_index < self.loader_batch_size:
										while self._loader_batch_index < self.loader_batch_size:
													lowerCamelCase			      =					self.loader_batch_item()
													lowerCamelCase			      =					item.pop("""is_last""" )
													accumulator.append(_a )
													if is_last:
																return accumulator

							while not is_last:
										lowerCamelCase			      =					self.infer(next(self.iterator )		, **self.params )
										if self.loader_batch_size is not None:
													if isinstance(_a		, torch.Tensor ):
																lowerCamelCase			      =					processed
													else:
																lowerCamelCase			      =					list(processed.keys() )[0]
																lowerCamelCase			      =					processed[key]
													if isinstance(_a		, _a ):
																lowerCamelCase			      =					len(_a )
													else:
																lowerCamelCase			      =					first_tensor.shape[0]
													if 0 < observed_batch_size < self.loader_batch_size:
																# could be last batch so we can't unroll as many
																# elements.
																lowerCamelCase			      =					observed_batch_size
													lowerCamelCase			      =					processed
													lowerCamelCase			      =					0
													while self._loader_batch_index < self.loader_batch_size:
																lowerCamelCase			      =					self.loader_batch_item()
																lowerCamelCase			      =					item.pop("""is_last""" )
																accumulator.append(_a )
																if is_last:
																			return accumulator
										else:
													lowerCamelCase			      =					processed
													lowerCamelCase			      =					item.pop("""is_last""" )
													accumulator.append(_a )
							return accumulator






class     __magic_name__							(       UpperCAmelCase__				):






				'''simple docstring'''





				def __init__(       self		, _a		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					dataset
							lowerCamelCase			      =					key





				def __len__(       self ):





							"""simple docstring"""
							return len(self.dataset )





				def __getitem__(       self		, _a ):





							"""simple docstring"""
							return self.dataset[i][self.key]






class     __magic_name__							(       UpperCAmelCase__				):






				'''simple docstring'''





				def __init__(       self		, _a		, _a		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					dataset
							lowerCamelCase			      =					keya
							lowerCamelCase			      =					keya





				def __len__(       self ):





							"""simple docstring"""
							return len(self.dataset )





				def __getitem__(       self		, _a ):





							"""simple docstring"""
							return {"text": self.dataset[i][self.keya], "text_pair": self.dataset[i][self.keya]}



