





import gc
import random
import unittest

import numpy as np
import torch

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetPipeline,
    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   _UpperCamelCase  (       A_						,       unittest.TestCase  ):







					'''simple docstring'''

					lowerCamelCase     :     Tuple			    =       KandinskyVaaControlnetPipeline
					lowerCamelCase     :     int			    =       ['image_embeds', 'negative_image_embeds', 'hint']
					lowerCamelCase     :     Union[str, Any]			    =       ['image_embeds', 'negative_image_embeds', 'hint']
					lowerCamelCase     :     Tuple			    =       [
					    'generator',
					    'height',
					    'width',
					    'latents',
					    'guidance_scale',
					    'num_inference_steps',
					    'return_dict',
					    'guidance_scale',
					    'num_images_per_prompt',
					    'output_type',
					    'return_dict',
					]
					lowerCamelCase     :     int			    =       False





					@property
					def        SCREAMING_SNAKE_CASE							(						self    :  Optional[int]		):






						'''simple docstring'''







						return 32





					@property
					def        SCREAMING_SNAKE_CASE							(						self    :  Any		):






						'''simple docstring'''







						return 32





					@property
					def        SCREAMING_SNAKE_CASE							(						self    :  Union[str, Any]		):






						'''simple docstring'''







						return self.time_input_dim





					@property
					def        SCREAMING_SNAKE_CASE							(						self    :  Optional[int]		):






						'''simple docstring'''







						return self.time_input_dim * 4





					@property
					def        SCREAMING_SNAKE_CASE							(						self    :  Optional[Any]		):






						'''simple docstring'''







						return 1_00





					@property
					def        SCREAMING_SNAKE_CASE							(						self    :  Dict		):






						'''simple docstring'''







						torch.manual_seed(0		)

						UpperCAmelCase_								=     {
						    """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,
						}

						UpperCAmelCase_								=     UNetaDConditionModel(**__lowercase		)
						return model





					@property
					def        SCREAMING_SNAKE_CASE							(						self    :  Optional[int]		):






						'''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        SCREAMING_SNAKE_CASE							(						self    :  str		):






						'''simple docstring'''







						torch.manual_seed(0		)
						UpperCAmelCase_								=     VQModel(**self.dummy_movq_kwargs		)
						return model





					def        SCREAMING_SNAKE_CASE							(						self    :  Union[str, Any]		):






						'''simple docstring'''







						UpperCAmelCase_								=     self.dummy_unet
						UpperCAmelCase_								=     self.dummy_movq

						UpperCAmelCase_								=     DDIMScheduler(
						    num_train_timesteps=10_00 ,					beta_schedule="""linear""" ,					beta_start=0.00085 ,					beta_end=0.012 ,					clip_sample=__lowercase ,					set_alpha_to_one=__lowercase ,					steps_offset=1 ,					prediction_type="""epsilon""" ,					thresholding=__lowercase ,					)

						UpperCAmelCase_								=     {
						    """unet""": unet,
						    """scheduler""": scheduler,
						    """movq""": movq,
						}
						return components





					def        SCREAMING_SNAKE_CASE							(						self    :  List[str] ,					__lowercase    :  Optional[int] ,					__lowercase    :  str=0		):






						'''simple docstring'''







						UpperCAmelCase_								=     floats_tensor((1, self.text_embedder_hidden_size) ,					rng=random.Random(__lowercase		)		).to(__lowercase		)
						UpperCAmelCase_								=     floats_tensor((1, self.text_embedder_hidden_size) ,					rng=random.Random(seed + 1		)		).to(
						    __lowercase		)

						# create hint
						UpperCAmelCase_								=     floats_tensor((1, 3, 64, 64) ,					rng=random.Random(__lowercase		)		).to(__lowercase		)

						if str(__lowercase		).startswith("""mps"""		):
							UpperCAmelCase_								=     torch.manual_seed(__lowercase		)
						else:
							UpperCAmelCase_								=     torch.Generator(device=__lowercase		).manual_seed(__lowercase		)
						UpperCAmelCase_								=     {
						    """image_embeds""": image_embeds,
						    """negative_image_embeds""": negative_image_embeds,
						    """hint""": hint,
						    """generator""": generator,
						    """height""": 64,
						    """width""": 64,
						    """guidance_scale""": 4.0,
						    """num_inference_steps""": 2,
						    """output_type""": """np""",
						}
						return inputs





					def        SCREAMING_SNAKE_CASE							(						self    :  int		):






						'''simple docstring'''







						UpperCAmelCase_								=     """cpu"""

						UpperCAmelCase_								=     self.get_dummy_components()

						UpperCAmelCase_								=     self.pipeline_class(**__lowercase		)
						UpperCAmelCase_								=     pipe.to(__lowercase		)

						pipe.set_progress_bar_config(disable=__lowercase		)

						UpperCAmelCase_								=     pipe(**self.get_dummy_inputs(__lowercase		)		)
						UpperCAmelCase_								=     output.images

						UpperCAmelCase_								=     pipe(
						    **self.get_dummy_inputs(__lowercase		) ,					return_dict=__lowercase ,					)[0]

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

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

						UpperCAmelCase_								=     np.array(
						    [0.6959826, 0.868279, 0.7558092, 0.68769467, 0.85805804, 0.65977496, 0.44885302, 0.5959111, 0.4251595]		)

						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   _UpperCamelCase  (       unittest.TestCase  ):







					'''simple docstring'''

					def        SCREAMING_SNAKE_CASE							(						self    :  int		):






						'''simple docstring'''







						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()





					def        SCREAMING_SNAKE_CASE							(						self    :  Union[str, Any]		):






						'''simple docstring'''







						UpperCAmelCase_								=     load_numpy(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
						    """/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy"""		)

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

						UpperCAmelCase_								=     KandinskyVaaPriorPipeline.from_pretrained(
						    """kandinsky-community/kandinsky-2-2-prior""" ,					torch_dtype=torch.floataa		)
						pipe_prior.to(__lowercase		)

						UpperCAmelCase_								=     KandinskyVaaControlnetPipeline.from_pretrained(
						    """kandinsky-community/kandinsky-2-2-controlnet-depth""" ,					torch_dtype=torch.floataa		)
						UpperCAmelCase_								=     pipeline.to(__lowercase		)
						pipeline.set_progress_bar_config(disable=__lowercase		)

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

						UpperCAmelCase_								=     torch.Generator(device="""cuda"""		).manual_seed(0		)
						UpperCAmelCase_						,	UpperCAmelCase_								=     pipe_prior(
						    __lowercase ,					generator=__lowercase ,					num_inference_steps=5 ,					negative_prompt="""""" ,					).to_tuple()

						UpperCAmelCase_								=     torch.Generator(device="""cuda"""		).manual_seed(0		)
						UpperCAmelCase_								=     pipeline(
						    image_embeds=__lowercase ,					negative_image_embeds=__lowercase ,					hint=__lowercase ,					generator=__lowercase ,					num_inference_steps=1_00 ,					output_type="""np""" ,					)

						UpperCAmelCase_								=     output.images[0]

						assert image.shape == (5_12, 5_12, 3)

						assert_mean_pixel_difference(__lowercase ,					__lowercase		)









from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


UpperCamelCase__   : List[str]					  = logging.get_logger(__name__)







class   _UpperCamelCase  (       A_  ):







					'''simple docstring'''

					def __init__(						self    :  Union[str, Any] ,					__lowercase    :  int ,					__lowercase    :  int ,					__lowercase    :  float ,					**__lowercase    :  Dict		):






						'''simple docstring'''







						UpperCAmelCase_								=     feature_size
						UpperCAmelCase_								=     sampling_rate
						UpperCAmelCase_								=     padding_value

						UpperCAmelCase_								=     kwargs.pop("""padding_side""" ,					"""right"""		)
						UpperCAmelCase_								=     kwargs.pop("""return_attention_mask""" ,					__lowercase		)

						super().__init__(**__lowercase		)





					def        SCREAMING_SNAKE_CASE							(						self    :  List[str] ,					__lowercase    :  Union[
					        BatchFeature,
					        List[BatchFeature],
					        Dict[str, BatchFeature],
					        Dict[str, List[BatchFeature]],
					        List[Dict[str, BatchFeature]],
					    ] ,					__lowercase    :  Union[bool, str, PaddingStrategy] = True ,					__lowercase    :  Optional[int] = None ,					__lowercase    :  bool = False ,					__lowercase    :  Optional[int] = None ,					__lowercase    :  Optional[bool] = None ,					__lowercase    :  Optional[Union[str, TensorType]] = None ,					):






						'''simple docstring'''







						if isinstance(__lowercase ,					(list, tuple)		) and isinstance(processed_features[0] ,					(dict, BatchFeature)		):
							UpperCAmelCase_								=     {
							    key: [example[key] for example in processed_features] for key in processed_features[0].keys()
							}

						# The model's main input name, usually `input_values`, has be passed for padding
						if self.model_input_names[0] not in processed_features:
							raise ValueError(
							    """You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`"""
							    F""" to this method that includes {self.model_input_names[0]}, but you provided"""
							    F""" {list(processed_features.keys()		)}"""		)

						UpperCAmelCase_								=     processed_features[self.model_input_names[0]]
						UpperCAmelCase_								=     (
						    return_attention_mask if return_attention_mask is not None else self.return_attention_mask
						)

						if len(__lowercase		) == 0:
							if return_attention_mask:
								UpperCAmelCase_								=     []
							return processed_features

						# If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
						# and rebuild them afterwards if no return_tensors is specified
						# Note that we lose the specific device the tensor may be on for PyTorch

						UpperCAmelCase_								=     required_input[0]
						if isinstance(__lowercase ,					(list, tuple)		):
							# first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
							UpperCAmelCase_								=     0
							while len(required_input[index]		) == 0:
								index += 1
							if index < len(__lowercase		):
								UpperCAmelCase_								=     required_input[index][0]

						if return_tensors is None:
							if is_tf_tensor(__lowercase		):
								UpperCAmelCase_								=     """tf"""
							elif is_torch_tensor(__lowercase		):
								UpperCAmelCase_								=     """pt"""
							elif isinstance(__lowercase ,					(int, float, list, tuple, np.ndarray)		):
								UpperCAmelCase_								=     """np"""
							else:
								raise ValueError(
								    F"""type of {first_element} unknown: {type(__lowercase		)}. """
								    """Should be one of a python, numpy, pytorch or tensorflow object."""		)

						for key, value in processed_features.items():
							if isinstance(value[0] ,					(int, float)		):
								UpperCAmelCase_								=     to_numpy(__lowercase		)
							else:
								UpperCAmelCase_								=     [to_numpy(__lowercase		) for v in value]

        # Convert padding_strategy in PaddingStrategy
						UpperCAmelCase_								=     self._get_padding_strategies(padding=__lowercase ,					max_length=__lowercase		)

						UpperCAmelCase_								=     processed_features[self.model_input_names[0]]

						UpperCAmelCase_								=     len(__lowercase		)
						if not all(len(__lowercase		) == batch_size for v in processed_features.values()		):
							raise ValueError("""Some items in the output dictionary have a different batch size than others."""		)

						UpperCAmelCase_								=     []
						for i in range(__lowercase		):
							UpperCAmelCase_								=     {k: v[i] for k, v in processed_features.items()}
							# truncation
							UpperCAmelCase_								=     self._truncate(
							    __lowercase ,					max_length=__lowercase ,					pad_to_multiple_of=__lowercase ,					truncation=__lowercase ,					)
							truncated_inputs.append(__lowercase		)

						if padding_strategy == PaddingStrategy.LONGEST:
							# make sure that `max_length` cannot be longer than the longest truncated length
							UpperCAmelCase_								=     max(len(input_slice[self.model_input_names[0]]		) for input_slice in truncated_inputs		)
							UpperCAmelCase_								=     PaddingStrategy.MAX_LENGTH

						UpperCAmelCase_								=     {}
						for i in range(__lowercase		):
							# padding
							UpperCAmelCase_								=     self._pad(
							    truncated_inputs[i] ,					max_length=__lowercase ,					padding_strategy=__lowercase ,					pad_to_multiple_of=__lowercase ,					return_attention_mask=__lowercase ,					)

							for key, value in outputs.items():
								if key not in batch_outputs:
									UpperCAmelCase_								=     []
								if value.dtype is np.dtype(np.floataa		):
									UpperCAmelCase_								=     value.astype(np.floataa		)
								batch_outputs[key].append(__lowercase		)

						return BatchFeature(__lowercase ,					tensor_type=__lowercase		)





					def        SCREAMING_SNAKE_CASE							(						self    :  str ,					__lowercase    :  Union[Dict[str, np.ndarray], BatchFeature] ,					__lowercase    :  Optional[int] = None ,					__lowercase    :  PaddingStrategy = PaddingStrategy.DO_NOT_PAD ,					__lowercase    :  Optional[int] = None ,					__lowercase    :  Optional[bool] = None ,					):






						'''simple docstring'''







						UpperCAmelCase_								=     processed_features[self.model_input_names[0]]

						if padding_strategy == PaddingStrategy.LONGEST:
							UpperCAmelCase_								=     len(__lowercase		)

						if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
							UpperCAmelCase_								=     ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

						UpperCAmelCase_								=     padding_strategy != PaddingStrategy.DO_NOT_PAD and len(__lowercase		) < max_length

						if return_attention_mask and "attention_mask" not in processed_features:
							UpperCAmelCase_								=     np.ones(len(__lowercase		) ,					dtype=np.intaa		)

						if needs_to_be_padded:
							UpperCAmelCase_								=     max_length - len(__lowercase		)
							if self.padding_side == "right":
								if return_attention_mask:
									UpperCAmelCase_								=     np.pad(
									    processed_features["""attention_mask"""] ,					(0, difference)		)
								UpperCAmelCase_								=     ((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
								UpperCAmelCase_								=     np.pad(
								    __lowercase ,					__lowercase ,					"""constant""" ,					constant_values=self.padding_value		)
							elif self.padding_side == "left":
								if return_attention_mask:
									UpperCAmelCase_								=     np.pad(
									    processed_features["""attention_mask"""] ,					(difference, 0)		)
								UpperCAmelCase_								=     ((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
								UpperCAmelCase_								=     np.pad(
								    __lowercase ,					__lowercase ,					"""constant""" ,					constant_values=self.padding_value		)
							else:
								raise ValueError("""Invalid padding strategy:""" + str(self.padding_side		)		)

						return processed_features





					def        SCREAMING_SNAKE_CASE							(						self    :  Optional[int] ,					__lowercase    :  Union[Dict[str, np.ndarray], BatchFeature] ,					__lowercase    :  Optional[int] = None ,					__lowercase    :  Optional[int] = None ,					__lowercase    :  Optional[bool] = None ,					):






						'''simple docstring'''







						if not truncation:
							return processed_features
						elif truncation and max_length is None:
							raise ValueError("""When setting ``truncation=True``, make sure that ``max_length`` is defined."""		)

						UpperCAmelCase_								=     processed_features[self.model_input_names[0]]

						# find `max_length` that fits `pad_to_multiple_of`
						if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
							UpperCAmelCase_								=     ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

						UpperCAmelCase_								=     len(__lowercase		) > max_length

						if needs_to_be_truncated:
							UpperCAmelCase_								=     processed_features[self.model_input_names[0]][:max_length]
							if "attention_mask" in processed_features:
								UpperCAmelCase_								=     processed_features["""attention_mask"""][:max_length]

						return processed_features





					def        SCREAMING_SNAKE_CASE							(						self    :  List[Any] ,					__lowercase    :  int=False ,					__lowercase    :  List[str]=None		):






						'''simple docstring'''







						if padding is not False:
							if padding is True:
								UpperCAmelCase_								=     PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
							elif not isinstance(__lowercase ,					__lowercase		):
								UpperCAmelCase_								=     PaddingStrategy(__lowercase		)
							elif isinstance(__lowercase ,					__lowercase		):
								UpperCAmelCase_								=     padding
						else:
							UpperCAmelCase_								=     PaddingStrategy.DO_NOT_PAD

						# Set max length if needed
						if max_length is None:
							if padding_strategy == PaddingStrategy.MAX_LENGTH:
								raise ValueError(
								    F"""When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined"""		)

        # Test if we have a padding value
						if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
							raise ValueError(
							    """Asking to pad but the feature_extractor does not have a padding value. Please select a value to use"""
							    """ as `padding_value`. For example: `feature_extractor.padding_value = 0.0`."""		)

						return padding_strategy


