


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        _a							(			UpperCamelCase__     ,	unittest.TestCase			):
							_lowercase	:			str        = KandinskyVaaControlnetImgaImgPipeline
							_lowercase	:			List[str]        = ['''image_embeds''', '''negative_image_embeds''', '''image''', '''hint''']
							_lowercase	:			Tuple        = ['''image_embeds''', '''negative_image_embeds''', '''image''', '''hint''']
							_lowercase	:			List[Any]        = [
							    '''generator''',
							    '''height''',
							    '''width''',
							    '''strength''',
							    '''guidance_scale''',
							    '''num_inference_steps''',
							    '''return_dict''',
							    '''guidance_scale''',
							    '''num_images_per_prompt''',
							    '''output_type''',
							    '''return_dict''',
							]
							_lowercase	:			Any        = False

							@property
							def       lowerCamelCase_ (							self:	Optional[Any]							)			->   Any:





									"""simple docstring"""







									return 32

							@property
							def       lowerCamelCase_ (							self:	List[str]							)			->   List[Any]:





									"""simple docstring"""







									return 32

							@property
							def       lowerCamelCase_ (							self:	Optional[int]							)			->   List[Any]:





									"""simple docstring"""







									return self.time_input_dim

							@property
							def       lowerCamelCase_ (							self:	Union[str, Any]							)			->   Optional[int]:





									"""simple docstring"""







									return self.time_input_dim * 4

							@property
							def       lowerCamelCase_ (							self:	Union[str, Any]							)			->   Tuple:





									"""simple docstring"""







									return 100

							@property
							def       lowerCamelCase_ (							self:	List[str]							)			->   Optional[int]:





									"""simple docstring"""







									torch.manual_seed(0							)

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

									lowercase__            =      UNetaDConditionModel(**UpperCamelCase_							)
									return model

							@property
							def       lowerCamelCase_ (							self:	Dict							)			->   Dict:





									"""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:	Optional[int]							)			->   str:





									"""simple docstring"""







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

							def       lowerCamelCase_ (							self:	Any							)			->   Optional[int]:





									"""simple docstring"""







									lowercase__            =      self.dummy_unet
									lowercase__            =      self.dummy_movq

									lowercase__            =      {
									    '''num_train_timesteps''': 1_000,
									    '''beta_schedule''': '''linear''',
									    '''beta_start''': 0.00085,
									    '''beta_end''': 0.012,
									    '''clip_sample''': False,
									    '''set_alpha_to_one''': False,
									    '''steps_offset''': 0,
									    '''prediction_type''': '''epsilon''',
									    '''thresholding''': False,
									}

									lowercase__            =      DDIMScheduler(**UpperCamelCase_							)

									lowercase__            =      {
									    '''unet''': unet,
									    '''scheduler''': scheduler,
									    '''movq''': movq,
									}

									return components

							def       lowerCamelCase_ (							self:	int    ,						UpperCamelCase_:	Tuple    ,						UpperCamelCase_:	List[str]=0							)			->   Optional[Any]:





									"""simple docstring"""







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

									if str(UpperCamelCase_							).startswith('''mps'''							):
											lowercase__            =      torch.manual_seed(UpperCamelCase_							)
									else:
											lowercase__            =      torch.Generator(device=UpperCamelCase_							).manual_seed(UpperCamelCase_							)
									lowercase__            =      {
									    '''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:	Union[str, Any]							)			->   int:





									"""simple docstring"""







									lowercase__            =      '''cpu'''

									lowercase__            =      self.get_dummy_components()

									lowercase__            =      self.pipeline_class(**UpperCamelCase_							)
									lowercase__            =      pipe.to(UpperCamelCase_							)

									pipe.set_progress_bar_config(disable=UpperCamelCase_							)

									lowercase__            =      pipe(**self.get_dummy_inputs(UpperCamelCase_							)							)
									lowercase__            =      output.images

									lowercase__            =      pipe(
									    **self.get_dummy_inputs(UpperCamelCase_							)    ,						return_dict=UpperCamelCase_    ,						)[0]

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

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

									lowercase__            =      np.array(
									    [0.54985034, 0.55509365, 0.52561504, 0.5570494, 0.5593818, 0.5263979, 0.50285643, 0.5069846, 0.51196736]							)
									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        _a							(			unittest.TestCase			):

							def       lowerCamelCase_ (							self:	Dict							)			->   int:





									"""simple docstring"""







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


							def       lowerCamelCase_ (							self:	Optional[Any]							)			->   Optional[Any]:





									"""simple docstring"""







									lowercase__            =      load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
									    '''/kandinskyv22/kandinskyv22_controlnet_img2img_robotcat_fp16.npy'''							)

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

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

									lowercase__            =      '''A robot, 4k photo'''

									lowercase__            =      KandinskyVaaPriorEmbaEmbPipeline.from_pretrained(
									    '''kandinsky-community/kandinsky-2-2-prior'''    ,						torch_dtype=torch.floataa							)
									pipe_prior.to(UpperCamelCase_							)

									lowercase__            =      KandinskyVaaControlnetImgaImgPipeline.from_pretrained(
									    '''kandinsky-community/kandinsky-2-2-controlnet-depth'''    ,						torch_dtype=torch.floataa							)
									lowercase__            =      pipeline.to(UpperCamelCase_							)

									pipeline.set_progress_bar_config(disable=UpperCamelCase_							)

									lowercase__            =      torch.Generator(device='''cpu'''							).manual_seed(0							)

									lowercase__     ,		lowercase__            =      pipe_prior(
									    UpperCamelCase_    ,						image=UpperCamelCase_    ,						strength=0.85    ,						generator=UpperCamelCase_    ,						negative_prompt=''''''    ,						).to_tuple()

									lowercase__            =      pipeline(
									    image=UpperCamelCase_    ,						image_embeds=UpperCamelCase_    ,						negative_image_embeds=UpperCamelCase_    ,						hint=UpperCamelCase_    ,						generator=UpperCamelCase_    ,						num_inference_steps=100    ,						height=512    ,						width=512    ,						strength=0.5    ,						output_type='''np'''    ,						)

									lowercase__            =      output.images[0]

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

									assert_mean_pixel_difference(UpperCamelCase_    ,						UpperCamelCase_							)






from pickle import UnpicklingError

import jax
import jax.numpy as jnp
import numpy as np
from flax.serialization import from_bytes
from flax.traverse_util import flatten_dict

from ..utils import logging


lowerCAmelCase       =							logging.get_logger(__name__)
def 			_a (      SCREAMING_SNAKE_CASE						,				SCREAMING_SNAKE_CASE     ):


		"""simple docstring"""




		try:
				with open(SCREAMING_SNAKE_CASE						,				'''rb'''     ) as flax_state_f:
						lowercase__            =      from_bytes(SCREAMING_SNAKE_CASE						,				flax_state_f.read()     )
		except UnpicklingError as e:
				try:
						with open(SCREAMING_SNAKE_CASE     ) as f:
								if f.read().startswith('''version'''     ):
										raise OSError(
										    '''You seem to have cloned a repository without having git-lfs installed. Please'''
										    ''' install git-lfs and run `git lfs install` followed by `git lfs pull` in the'''
										    ''' folder you cloned.'''     )
								else:
										raise ValueError from e
				except (UnicodeDecodeError, ValueError):
						raise EnvironmentError(f'Unable to convert {model_file} to Flax deserializable object. '     )

		return load_flax_weights_in_pytorch_model(SCREAMING_SNAKE_CASE						,				SCREAMING_SNAKE_CASE     )
def 			_a (      SCREAMING_SNAKE_CASE						,				SCREAMING_SNAKE_CASE     ):


		"""simple docstring"""




		try:
				import torch  # noqa: F401
		except ImportError:
				logger.error(
				    '''Loading Flax weights in PyTorch requires both PyTorch and Flax to be installed. Please see'''
				    ''' https://pytorch.org/ and https://flax.readthedocs.io/en/latest/installation.html for installation'''
				    ''' instructions.'''     )
				raise

		# check if we have bf16 weights
		lowercase__            =      flatten_dict(jax.tree_util.tree_map(lambda SCREAMING_SNAKE_CASE     : x.dtype == jnp.bfloataa						,				SCREAMING_SNAKE_CASE     )     ).values()
		if any(SCREAMING_SNAKE_CASE     ):
				# convert all weights to fp32 if they are bf16 since torch.from_numpy can-not handle bf16

				# and bf16 is not fully supported in PT yet.
				logger.warning(
				    '''Found ``bfloat16`` weights in Flax model. Casting all ``bfloat16`` weights to ``float32`` '''
				    '''before loading those in PyTorch model.'''     )
				lowercase__            =      jax.tree_util.tree_map(
				    lambda SCREAMING_SNAKE_CASE     : params.astype(np.floataa     ) if params.dtype == jnp.bfloataa else params						,				SCREAMING_SNAKE_CASE     )

		lowercase__            =      ''''''

		lowercase__            =      flatten_dict(SCREAMING_SNAKE_CASE						,				sep='''.'''     )
		lowercase__            =      pt_model.state_dict()

		# keep track of unexpected & missing keys
		lowercase__            =      []
		lowercase__            =      set(pt_model_dict.keys()     )

		for flax_key_tuple, flax_tensor in flax_state_dict.items():
				lowercase__            =      flax_key_tuple.split('''.'''     )

				if flax_key_tuple_array[-1] == "kernel" and flax_tensor.ndim == 4:
						lowercase__            =      flax_key_tuple_array[:-1] + ['''weight''']
						lowercase__            =      jnp.transpose(SCREAMING_SNAKE_CASE						,				(3, 2, 0, 1)     )
				elif flax_key_tuple_array[-1] == "kernel":
						lowercase__            =      flax_key_tuple_array[:-1] + ['''weight''']
						lowercase__            =      flax_tensor.T
				elif flax_key_tuple_array[-1] == "scale":
						lowercase__            =      flax_key_tuple_array[:-1] + ['''weight''']

				if "time_embedding" not in flax_key_tuple_array:
						for i, flax_key_tuple_string in enumerate(SCREAMING_SNAKE_CASE     ):
								lowercase__            =      (
								    flax_key_tuple_string.replace('''_0'''						,				'''.0'''     )
								    .replace('''_1'''						,				'''.1'''     )
								    .replace('''_2'''						,				'''.2'''     )
								    .replace('''_3'''						,				'''.3'''     )
								    .replace('''_4'''						,				'''.4'''     )
								    .replace('''_5'''						,				'''.5'''     )
								    .replace('''_6'''						,				'''.6'''     )
								    .replace('''_7'''						,				'''.7'''     )
								    .replace('''_8'''						,				'''.8'''     )
								    .replace('''_9'''						,				'''.9'''     )
								)

				lowercase__            =      '''.'''.join(SCREAMING_SNAKE_CASE     )

				if flax_key in pt_model_dict:
						if flax_tensor.shape != pt_model_dict[flax_key].shape:
								raise ValueError(
								    f'Flax checkpoint seems to be incorrect. Weight {flax_key_tuple} was expected '
								    f'to be of shape {pt_model_dict[flax_key].shape}, but is {flax_tensor.shape}.'     )
						else:
								# add weight to pytorch dict
								lowercase__            =      np.asarray(SCREAMING_SNAKE_CASE     ) if not isinstance(SCREAMING_SNAKE_CASE						,				np.ndarray     ) else flax_tensor
								lowercase__            =      torch.from_numpy(SCREAMING_SNAKE_CASE     )
								# remove from missing keys
								missing_keys.remove(SCREAMING_SNAKE_CASE     )
				else:
						# weight is not expected by PyTorch model
						unexpected_keys.append(SCREAMING_SNAKE_CASE     )

		pt_model.load_state_dict(SCREAMING_SNAKE_CASE     )

		# re-transform missing_keys to list
		lowercase__            =      list(SCREAMING_SNAKE_CASE     )

		if len(SCREAMING_SNAKE_CASE     ) > 0:
				logger.warning(
				    '''Some weights of the Flax model were not used when initializing the PyTorch model'''
				    f' {pt_model.__class__.__name__}: {unexpected_keys}\n- This IS expected if you are initializing'
				    f' {pt_model.__class__.__name__} from a Flax model trained on another task or with another architecture'
				    ''' (e.g. initializing a BertForSequenceClassification model from a FlaxBertForPreTraining model).\n- This'''
				    f' IS NOT expected if you are initializing {pt_model.__class__.__name__} from a Flax model that you expect'
				    ''' to be exactly identical (e.g. initializing a BertForSequenceClassification model from a'''
				    ''' FlaxBertForSequenceClassification model).'''     )
		if len(SCREAMING_SNAKE_CASE     ) > 0:
				logger.warning(
				    f'Some weights of {pt_model.__class__.__name__} were not initialized from the Flax model and are newly'
				    f' initialized: {missing_keys}\nYou should probably TRAIN this model on a down-stream task to be able to'
				    ''' use it for predictions and inference.'''     )

		return pt_model


