





'''simple docstring'''






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 		lowerCamelCase	(   _A   , unittest.TestCase							):
	snake_case_   = KandinskyVaaControlnetPipeline
	snake_case_   = ["image_embeds", "negative_image_embeds", "hint"]
	snake_case_   = ["image_embeds", "negative_image_embeds", "hint"]
	snake_case_   = [
	    "generator",
	    "height",
	    "width",
	    "latents",
	    "guidance_scale",
	    "num_inference_steps",
	    "return_dict",
	    "guidance_scale",
	    "num_images_per_prompt",
	    "output_type",
	    "return_dict",
	]
	snake_case_   = False



	@property
	def 	_lowerCamelCase     ( self						):
			return 32



	@property
	def 	_lowerCamelCase     ( self						):
			return 32



	@property
	def 	_lowerCamelCase     ( self						):
			return self.time_input_dim



	@property
	def 	_lowerCamelCase     ( self						):
			return self.time_input_dim * 4



	@property
	def 	_lowerCamelCase     ( self						):
			return 100



	@property
	def 	_lowerCamelCase     ( self						):
			torch.manual_seed(0						)

			lowerCAmelCase  :     str       						= {
			    "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  :     Union[str, Any]       						= UNetaDConditionModel(**a_						)
			return model



	@property
	def 	_lowerCamelCase     ( self						):
			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						):
			torch.manual_seed(0						)
			lowerCAmelCase  :     Tuple       						= VQModel(**self.dummy_movq_kwargs						)
			return model



	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     Tuple       						= self.dummy_unet
			lowerCAmelCase  :     Optional[int]       						= self.dummy_movq

			lowerCAmelCase  :     Any       						= DDIMScheduler(
			    num_train_timesteps=1_000		,   beta_schedule="linear"		,   beta_start=0.00085		,   beta_end=0.012		,   clip_sample=a_		,   set_alpha_to_one=a_		,   steps_offset=1		,   prediction_type="epsilon"		,   thresholding=a_		,   )

			lowerCAmelCase  :     Union[str, Any]       						= {
			    "unet": unet,
			    "scheduler": scheduler,
			    "movq": movq,
			}
			return components



	def 	_lowerCamelCase     ( self		,   a_		,   a_=0						):
			lowerCAmelCase  :     Optional[int]       						= floats_tensor((1, self.text_embedder_hidden_size)		,   rng=random.Random(a_						)						).to(a_						)
			lowerCAmelCase  :     Any       						= floats_tensor((1, self.text_embedder_hidden_size)		,   rng=random.Random(seed + 1						)						).to(
			    a_						)

			# create hint
			lowerCAmelCase  :     int       						= floats_tensor((1, 3, 64, 64)		,   rng=random.Random(a_						)						).to(a_						)

			if str(a_						).startswith("mps"						):
					lowerCAmelCase  :     Optional[int]       						= torch.manual_seed(a_						)
			else:
					lowerCAmelCase  :     Optional[Any]       						= torch.Generator(device=a_						).manual_seed(a_						)
			lowerCAmelCase  :     Dict       						= {
			    "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 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     Any       						= "cpu"

			lowerCAmelCase  :     int       						= self.get_dummy_components()

			lowerCAmelCase  :     Any       						= self.pipeline_class(**a_						)
			lowerCAmelCase  :     Dict       						= pipe.to(a_						)

			pipe.set_progress_bar_config(disable=a_						)

			lowerCAmelCase  :     Optional[Any]       						= pipe(**self.get_dummy_inputs(a_						)						)
			lowerCAmelCase  :     int       						= output.images

			lowerCAmelCase  :     Any       						= pipe(
			    **self.get_dummy_inputs(a_						)		,   return_dict=a_		,   )[0]

			lowerCAmelCase  :     List[Any]       						= image[0, -3:, -3:, -1]
			lowerCAmelCase  :     Dict       						= image_from_tuple[0, -3:, -3:, -1]

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

			lowerCAmelCase  :     Optional[Any]       						= 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 		lowerCamelCase	(   unittest.TestCase							):



	def 	_lowerCamelCase     ( self						):
			# clean up the VRAM after each test
			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()





	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     Optional[int]       						= load_numpy(
			    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
			    "/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy"						)

			lowerCAmelCase  :     int       						= load_image(
			    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
			    "/kandinskyv22/hint_image_cat.png"						)
			lowerCAmelCase  :     Optional[Any]       						= torch.from_numpy(np.array(a_						)						).float() / 255.0
			lowerCAmelCase  :     Union[str, Any]       						= hint.permute(2		,   0		,   1						).unsqueeze(0						)

			lowerCAmelCase  :     Any       						= KandinskyVaaPriorPipeline.from_pretrained(
			    "kandinsky-community/kandinsky-2-2-prior"		,   torch_dtype=torch.floataa						)
			pipe_prior.to(a_						)

			lowerCAmelCase  :     str       						= KandinskyVaaControlnetPipeline.from_pretrained(
			    "kandinsky-community/kandinsky-2-2-controlnet-depth"		,   torch_dtype=torch.floataa						)
			lowerCAmelCase  :     Dict       						= pipeline.to(a_						)
			pipeline.set_progress_bar_config(disable=a_						)

			lowerCAmelCase  :     Optional[int]       						= "A robot, 4k photo"

			lowerCAmelCase  :     Optional[int]       						= torch.Generator(device="cuda"						).manual_seed(0						)
			lowerCAmelCase						,		lowerCAmelCase  :     Optional[int]       						= pipe_prior(
			    a_		,   generator=a_		,   num_inference_steps=5		,   negative_prompt=""		,   ).to_tuple()

			lowerCAmelCase  :     List[str]       						= torch.Generator(device="cuda"						).manual_seed(0						)
			lowerCAmelCase  :     int       						= pipeline(
			    image_embeds=a_		,   negative_image_embeds=a_		,   hint=a_		,   generator=a_		,   num_inference_steps=100		,   output_type="np"		,   )

			lowerCAmelCase  :     Any       						= output.images[0]

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

			assert_mean_pixel_difference(a_		,   a_						)










'''simple docstring'''






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


lowerCAmelCase									=   logging.get_logger(__name__)
class 		lowerCamelCase	(   _A							):



	def __init__( self		,   a_		,   a_		,   a_		,   **a_						):
			lowerCAmelCase  :     Tuple       						= feature_size
			lowerCAmelCase  :     int       						= sampling_rate
			lowerCAmelCase  :     str       						= padding_value

			lowerCAmelCase  :     int       						= kwargs.pop("padding_side"		,   "right"						)
			lowerCAmelCase  :     int       						= kwargs.pop("return_attention_mask"		,   a_						)

			super().__init__(**a_						)



	def 	_lowerCamelCase     ( self		,   a_		,   a_ = True		,   a_ = None		,   a_ = False		,   a_ = None		,   a_ = None		,   a_ = None		,   ):
			# If we have a list of dicts, let's convert it in a dict of lists
			# We do this to allow using this method as a collate_fn function in PyTorch Dataloader
			if isinstance(a_		,   (list, tuple)						) and isinstance(processed_features[0]		,   (dict, BatchFeature)						):
					lowerCAmelCase  :     str       						= {
					    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()						)}'''						)

			lowerCAmelCase  :     Any       						= processed_features[self.model_input_names[0]]
			lowerCAmelCase  :     Dict       						= (
			    return_attention_mask if return_attention_mask is not None else self.return_attention_mask
			)

			if len(a_						) == 0:
					if return_attention_mask:
							lowerCAmelCase  :     List[Any]       						= []
					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

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

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

			for key, value in processed_features.items():
					if isinstance(value[0]		,   (int, float)						):
							lowerCAmelCase  :     Dict       						= to_numpy(a_						)
					else:
							lowerCAmelCase  :     Any       						= [to_numpy(a_						) for v in value]

        # Convert padding_strategy in PaddingStrategy
			lowerCAmelCase  :     Optional[int]       						= self._get_padding_strategies(padding=a_		,   max_length=a_						)

			lowerCAmelCase  :     Dict       						= processed_features[self.model_input_names[0]]

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

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

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

			lowerCAmelCase  :     List[Any]       						= {}
			for i in range(a_						):
					# padding
					lowerCAmelCase  :     Optional[int]       						= self._pad(
					    truncated_inputs[i]		,   max_length=a_		,   padding_strategy=a_		,   pad_to_multiple_of=a_		,   return_attention_mask=a_		,   )

					for key, value in outputs.items():
							if key not in batch_outputs:
									lowerCAmelCase  :     Dict       						= []
							if value.dtype is np.dtype(np.floataa						):
									lowerCAmelCase  :     int       						= value.astype(np.floataa						)
							batch_outputs[key].append(a_						)

			return BatchFeature(a_		,   tensor_type=a_						)



	def 	_lowerCamelCase     ( self		,   a_		,   a_ = None		,   a_ = PaddingStrategy.DO_NOT_PAD		,   a_ = None		,   a_ = None		,   ):
			lowerCAmelCase  :     Tuple       						= processed_features[self.model_input_names[0]]

			if padding_strategy == PaddingStrategy.LONGEST:
					lowerCAmelCase  :     Dict       						= len(a_						)

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

			lowerCAmelCase  :     List[Any]       						= padding_strategy != PaddingStrategy.DO_NOT_PAD and len(a_						) < max_length

			if return_attention_mask and "attention_mask" not in processed_features:
					lowerCAmelCase  :     Any       						= np.ones(len(a_						)		,   dtype=np.intaa						)

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

			return processed_features



	def 	_lowerCamelCase     ( self		,   a_		,   a_ = None		,   a_ = None		,   a_ = None		,   ):
			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."						)

			lowerCAmelCase  :     int       						= 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):
					lowerCAmelCase  :     Union[str, Any]       						= ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

			lowerCAmelCase  :     Any       						= len(a_						) > max_length

			if needs_to_be_truncated:
					lowerCAmelCase  :     Optional[int]       						= processed_features[self.model_input_names[0]][:max_length]
					if "attention_mask" in processed_features:
							lowerCAmelCase  :     Any       						= processed_features["attention_mask"][:max_length]

			return processed_features





	def 	_lowerCamelCase     ( self		,   a_=False		,   a_=None						):

			# Get padding strategy
			if padding is not False:
					if padding is True:
							lowerCAmelCase  :     str       						= PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
					elif not isinstance(a_		,   a_						):
							lowerCAmelCase  :     str       						= PaddingStrategy(a_						)
					elif isinstance(a_		,   a_						):
							lowerCAmelCase  :     Optional[int]       						= padding
			else:
					lowerCAmelCase  :     Tuple       						= 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



