


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       lowercase_					(lowercase__      ,					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 						__UpperCamelCase    (				self)  -> Any:
											return 32





				@property
				def 						__UpperCamelCase    (				self)  -> Union[str, Any]:
											return 32





				@property
				def 						__UpperCamelCase    (				self)  -> int:
											return self.time_input_dim





				@property
				def 						__UpperCamelCase    (				self)  -> Optional[Any]:
											return self.time_input_dim * 4





				@property
				def 						__UpperCamelCase    (				self)  -> Optional[int]:
											return 100





				@property
				def 						__UpperCamelCase    (				self)  -> Optional[Any]:
											torch.manual_seed(0)

											a__      ={
											    '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,
											}

											a__      =UNetaDConditionModel(**lowercase_)
											return model





				@property
				def 						__UpperCamelCase    (				self)  -> Tuple:
											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 						__UpperCamelCase    (				self)  -> Optional[Any]:
											torch.manual_seed(0)
											a__      =VQModel(**self.dummy_movq_kwargs)
											return model





				def 						__UpperCamelCase    (				self)  -> Union[str, Any]:
											a__      =self.dummy_unet
											a__      =self.dummy_movq

											a__      =DDIMScheduler(
											    num_train_timesteps=1000       ,  beta_schedule='linear'       ,  beta_start=0.0_00_85       ,  beta_end=0.0_12       ,  clip_sample=lowercase_       ,  set_alpha_to_one=lowercase_       ,  steps_offset=1       ,  prediction_type='epsilon'       ,  thresholding=lowercase_       ,  )

											a__      ={
											    'unet': unet,
											    'scheduler': scheduler,
											    'movq': movq,
											}
											return components





				def 						__UpperCamelCase    (				self       ,  lowercase_       ,  lowercase_=0)  -> Optional[int]:
											a__      =floats_tensor((1, self.text_embedder_hidden_size)       ,  rng=random.Random(lowercase_)).to(lowercase_)
											a__      =floats_tensor((1, self.text_embedder_hidden_size)       ,  rng=random.Random(seed + 1)).to(
											    lowercase_)

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

											if str(lowercase_).startswith('mps'):
																		a__      =torch.manual_seed(lowercase_)
											else:
																		a__      =torch.Generator(device=lowercase_).manual_seed(lowercase_)
											a__      ={
											    '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 						__UpperCamelCase    (				self)  -> Dict:
											a__      ='cpu'

											a__      =self.get_dummy_components()

											a__      =self.pipeline_class(**lowercase_)
											a__      =pipe.to(lowercase_)

											pipe.set_progress_bar_config(disable=lowercase_)

											a__      =pipe(**self.get_dummy_inputs(lowercase_))
											a__      =output.images

											a__      =pipe(
											    **self.get_dummy_inputs(lowercase_)       ,  return_dict=lowercase_       ,  )[0]

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

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

											a__      =np.array(
											    [0.6_95_98_26, 0.86_82_79, 0.7_55_80_92, 0.68_76_94_67, 0.85_80_58_04, 0.65_97_74_96, 0.44_88_53_02, 0.5_95_91_11, 0.4_25_15_95])

											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   ):





				def 						__UpperCamelCase    (				self)  -> Union[str, Any]:
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()





				def 						__UpperCamelCase    (				self)  -> Dict:
											a__      =load_numpy(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
											    '/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy')

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

											a__      =KandinskyVaaPriorPipeline.from_pretrained(
											    'kandinsky-community/kandinsky-2-2-prior'       ,  torch_dtype=torch.floataa)
											pipe_prior.to(lowercase_)

											a__      =KandinskyVaaControlnetPipeline.from_pretrained(
											    'kandinsky-community/kandinsky-2-2-controlnet-depth'       ,  torch_dtype=torch.floataa)
											a__      =pipeline.to(lowercase_)
											pipeline.set_progress_bar_config(disable=lowercase_)

											a__      ='A robot, 4k photo'

											a__      =torch.Generator(device='cuda').manual_seed(0)
											a__    ,     a__      =pipe_prior(
											    lowercase_       ,  generator=lowercase_       ,  num_inference_steps=5       ,  negative_prompt=''       ,  ).to_tuple()

											a__      =torch.Generator(device='cuda').manual_seed(0)
											a__      =pipeline(
											    image_embeds=lowercase_       ,  negative_image_embeds=lowercase_       ,  hint=lowercase_       ,  generator=lowercase_       ,  num_inference_steps=100       ,  output_type='np'       ,  )

											a__      =output.images[0]

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

											assert_mean_pixel_difference(lowercase_       ,  lowercase_)





from importlib import import_module

from .logging import get_logger


_lowerCAmelCase:						str   				=       get_logger(__name__)




class       lowercase_					:





				def __init__(				self       ,  lowercase_       ,  lowercase_=None)  -> Tuple:
											a__      =attrs or []
											if module is not None:
																		for key in module.__dict__:
																									if key in attrs or not key.startswith('__'):
																																setattr(self       ,  lowercase_       ,  getattr(lowercase_       ,  lowercase_))
											a__      =module._original_module if isinstance(lowercase_       ,  _PatchedModuleObj) else module







class       lowercase_					:
				snake_case	    =[]





				def __init__(				self       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_=None)  -> List[str]:
											a__      =obj
											a__      =target
											a__      =new
											a__      =target.split('.')[0]
											a__      ={}
											a__      =attrs or []





				def __enter__(				self)  -> Optional[int]:
											*a__    ,     a__      =self.target.split('.')

											# Patch modules:
											# it's used to patch attributes of submodules like "os.path.join";
											# in this case we need to patch "os" and "os.path"

											for i in range(len(lowercase_)):
																		try:
																									a__      =import_module('.'.join(submodules[: i + 1]))
																		except ModuleNotFoundError:
																									continue
																		# We iterate over all the globals in self.obj in case we find "os" or "os.path"
																		for attr in self.obj.__dir__():
																									a__      =getattr(self.obj       ,  lowercase_)
																									# We don't check for the name of the global, but rather if its value *is* "os" or "os.path".
																									# This allows to patch renamed modules like "from os import path as ospath".
																									if obj_attr is submodule or (
																									    (isinstance(lowercase_       ,  _PatchedModuleObj) and obj_attr._original_module is submodule)
																									):
																																a__      =obj_attr
																																# patch at top level
																																setattr(self.obj       ,  lowercase_       ,  _PatchedModuleObj(lowercase_       ,  attrs=self.attrs))
																																a__      =getattr(self.obj       ,  lowercase_)
																																# construct lower levels patches
																																for key in submodules[i + 1 :]:
																																							setattr(lowercase_       ,  lowercase_       ,  _PatchedModuleObj(getattr(lowercase_       ,  lowercase_       ,  lowercase_)       ,  attrs=self.attrs))
																																							a__      =getattr(lowercase_       ,  lowercase_)
																																# finally set the target attribute
																																setattr(lowercase_       ,  lowercase_       ,  self.new)

        # Patch attribute itself:
        # it's used for builtins like "open",
        # and also to patch "os.path.join" we may also need to patch "join"
        # itself if it was imported as "from os.path import join".

											if submodules:  # if it's an attribute of a submodule like "os.path.join"
																		try:
																									a__      =getattr(import_module('.'.join(lowercase_))       ,  lowercase_)
																		except (AttributeError, ModuleNotFoundError):
																									return
																		# We iterate over all the globals in self.obj in case we find "os.path.join"
																		for attr in self.obj.__dir__():
																									# We don't check for the name of the global, but rather if its value *is* "os.path.join".
																									# This allows to patch renamed attributes like "from os.path import join as pjoin".
																									if getattr(self.obj       ,  lowercase_) is attr_value:
																																a__      =getattr(self.obj       ,  lowercase_)
																																setattr(self.obj       ,  lowercase_       ,  self.new)
											elif target_attr in globals()["__builtins__"]:  # if it'a s builtin like "open"
																		a__      =globals()['__builtins__'][target_attr]
																		setattr(self.obj       ,  lowercase_       ,  self.new)
											else:
																		raise RuntimeError(F"""Tried to patch attribute {target_attr} instead of a submodule.""")





				def __exit__(				self       ,  *lowercase_)  -> str:
											for attr in list(self.original):
																		setattr(self.obj       ,  lowercase_       ,  self.original.pop(lowercase_))





				def 						__UpperCamelCase    (				self)  -> Any:
											self.__enter__()
											self._active_patches.append(self)





				def 						__UpperCamelCase    (				self)  -> Union[str, Any]:
											try:
																		self._active_patches.remove(self)
											except ValueError:
																		# If the patch hasn't been started this will fail
																		return None

											return self.__exit__()

