





'''simple docstring'''




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


a__					:			Any          =     logging.get_logger(__name__)




def 						__lowerCamelCase			(	UpperCAmelCase_					,				UpperCAmelCase_							)   ->Any:
	try:
		with open(UpperCAmelCase_					,				'rb'							) as flax_state_f:
			snake_case__	       =   from_bytes(UpperCAmelCase_					,				flax_state_f.read()							)
	except UnpicklingError as e:
		try:
			with open(UpperCAmelCase_							) 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(UpperCAmelCase_					,				UpperCAmelCase_							)







def 						__lowerCamelCase			(	UpperCAmelCase_					,				UpperCAmelCase_							)   ->Optional[int]:

	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
	snake_case__	       =   flatten_dict(jax.tree_util.tree_map(lambda UpperCAmelCase_							: x.dtype == jnp.bfloataa					,				UpperCAmelCase_							)							).values()
	if any(UpperCAmelCase_							):
		# 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.'							)
		snake_case__	       =   jax.tree_util.tree_map(
		    lambda UpperCAmelCase_							: params.astype(np.floataa							) if params.dtype == jnp.bfloataa else params					,				UpperCAmelCase_							)

	snake_case__	       =   ''

	snake_case__	       =   flatten_dict(UpperCAmelCase_					,				sep='.'							)
	snake_case__	       =   pt_model.state_dict()

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

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

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

		if "time_embedding" not in flax_key_tuple_array:
			for i, flax_key_tuple_string in enumerate(UpperCAmelCase_							):
				snake_case__	       =   (
				    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'							)
				)

		snake_case__	       =   '.'.join(UpperCAmelCase_							)

		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
				snake_case__	       =   np.asarray(UpperCAmelCase_							) if not isinstance(UpperCAmelCase_					,				np.ndarray							) else flax_tensor
				snake_case__	       =   torch.from_numpy(UpperCAmelCase_							)
				# remove from missing keys
				missing_keys.remove(UpperCAmelCase_							)
		else:
			# weight is not expected by PyTorch model
			unexpected_keys.append(UpperCAmelCase_							)

	pt_model.load_state_dict(UpperCAmelCase_							)

	# re-transform missing_keys to list
	snake_case__	       =   list(UpperCAmelCase_							)

	if len(UpperCAmelCase_							) > 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(UpperCAmelCase_							) > 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










'''simple docstring'''




import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableDiffusionUpscalePipeline, UNetaDConditionModel
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


enable_full_determinism()






class 				__snake_case     (					unittest.TestCase							):





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





	@property
	def   _snake_case   (		self		)      ->		str:
		snake_case__	       =   1
		snake_case__	       =   3
		snake_case__	       =   (32, 32)

		snake_case__	       =   floats_tensor((batch_size, num_channels) + sizes      ,       rng=random.Random(0		)		).to(UpperCamelCase_		)
		return image





	@property
	def   _snake_case   (		self		)      ->		Union[str, Any]:
		torch.manual_seed(0		)
		snake_case__	       =   UNetaDConditionModel(
		    block_out_channels=(32, 32, 64)      ,       layers_per_block=2      ,       sample_size=32      ,       in_channels=7      ,       out_channels=4      ,       down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D', 'CrossAttnDownBlock2D')      ,       up_block_types=('CrossAttnUpBlock2D', 'CrossAttnUpBlock2D', 'UpBlock2D')      ,       cross_attention_dim=32      ,       attention_head_dim=8      ,       use_linear_projection=UpperCamelCase_      ,       only_cross_attention=(True, True, False)      ,       num_class_embeds=100      ,       )
		return model





	@property
	def   _snake_case   (		self		)      ->		Optional[Any]:
		torch.manual_seed(0		)
		snake_case__	       =   AutoencoderKL(
		    block_out_channels=[32, 32, 64]      ,       in_channels=3      ,       out_channels=3      ,       down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D', 'DownEncoderBlock2D']      ,       up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D']      ,       latent_channels=4      ,       )
		return model





	@property
	def   _snake_case   (		self		)      ->		List[str]:
		torch.manual_seed(0		)
		snake_case__	       =   CLIPTextConfig(
		    bos_token_id=0      ,       eos_token_id=2      ,       hidden_size=32      ,       intermediate_size=37      ,       layer_norm_eps=1E-0_5      ,       num_attention_heads=4      ,       num_hidden_layers=5      ,       pad_token_id=1      ,       vocab_size=1000      ,       hidden_act='gelu'      ,       projection_dim=512      ,       )
		return CLIPTextModel(UpperCamelCase_		)





	def   _snake_case   (		self		)      ->		Union[str, Any]:
		snake_case__	       =   'cpu'  # ensure determinism for the device-dependent torch.Generator
		snake_case__	       =   self.dummy_cond_unet_upscale
		snake_case__	       =   DDPMScheduler()
		snake_case__	       =   DDIMScheduler(prediction_type='v_prediction'		)
		snake_case__	       =   self.dummy_vae
		snake_case__	       =   self.dummy_text_encoder
		snake_case__	       =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

		snake_case__	       =   self.dummy_image.cpu().permute(0      ,       2      ,       3      ,       1		)[0]
		snake_case__	       =   Image.fromarray(np.uinta(UpperCamelCase_		)		).convert('RGB'		).resize((64, 64)		)

		# make sure here that pndm scheduler skips prk
		snake_case__	       =   StableDiffusionUpscalePipeline(
		    unet=UpperCamelCase_      ,       low_res_scheduler=UpperCamelCase_      ,       scheduler=UpperCamelCase_      ,       vae=UpperCamelCase_      ,       text_encoder=UpperCamelCase_      ,       tokenizer=UpperCamelCase_      ,       max_noise_level=350      ,       )
		snake_case__	       =   sd_pipe.to(UpperCamelCase_		)
		sd_pipe.set_progress_bar_config(disable=UpperCamelCase_		)

		snake_case__	       =   'A painting of a squirrel eating a burger'
		snake_case__	       =   torch.Generator(device=UpperCamelCase_		).manual_seed(0		)
		snake_case__	       =   sd_pipe(
		    [prompt]      ,       image=UpperCamelCase_      ,       generator=UpperCamelCase_      ,       guidance_scale=6.0      ,       noise_level=20      ,       num_inference_steps=2      ,       output_type='np'      ,       )

		snake_case__	       =   output.images

		snake_case__	       =   torch.Generator(device=UpperCamelCase_		).manual_seed(0		)
		snake_case__	       =   sd_pipe(
		    [prompt]      ,       image=UpperCamelCase_      ,       generator=UpperCamelCase_      ,       guidance_scale=6.0      ,       noise_level=20      ,       num_inference_steps=2      ,       output_type='np'      ,       return_dict=UpperCamelCase_      ,       )[0]

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

		snake_case__	       =   low_res_image.size[0] * 4
		assert image.shape == (1, expected_height_width, expected_height_width, 3)
		snake_case__	       =   np.array([0.3_1_1_3, 0.3_9_1_0, 0.4_2_7_2, 0.4_8_5_9, 0.5_0_6_1, 0.4_6_5_2, 0.5_3_6_2, 0.5_7_1_5, 0.5_6_6_1]		)

		assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2
		assert np.abs(image_from_tuple_slice.flatten() - expected_slice		).max() < 1E-2





	def   _snake_case   (		self		)      ->		List[str]:
		snake_case__	       =   'cpu'  # ensure determinism for the device-dependent torch.Generator
		snake_case__	       =   self.dummy_cond_unet_upscale
		snake_case__	       =   DDPMScheduler()
		snake_case__	       =   DDIMScheduler(prediction_type='v_prediction'		)
		snake_case__	       =   self.dummy_vae
		snake_case__	       =   self.dummy_text_encoder
		snake_case__	       =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

		snake_case__	       =   self.dummy_image.cpu().permute(0      ,       2      ,       3      ,       1		)[0]
		snake_case__	       =   Image.fromarray(np.uinta(UpperCamelCase_		)		).convert('RGB'		).resize((64, 64)		)

		# make sure here that pndm scheduler skips prk
		snake_case__	       =   StableDiffusionUpscalePipeline(
		    unet=UpperCamelCase_      ,       low_res_scheduler=UpperCamelCase_      ,       scheduler=UpperCamelCase_      ,       vae=UpperCamelCase_      ,       text_encoder=UpperCamelCase_      ,       tokenizer=UpperCamelCase_      ,       max_noise_level=350      ,       )
		snake_case__	       =   sd_pipe.to(UpperCamelCase_		)
		sd_pipe.set_progress_bar_config(disable=UpperCamelCase_		)

		snake_case__	       =   'A painting of a squirrel eating a burger'
		snake_case__	       =   sd_pipe(
		    2 * [prompt]      ,       image=2 * [low_res_image]      ,       guidance_scale=6.0      ,       noise_level=20      ,       num_inference_steps=2      ,       output_type='np'      ,       )
		snake_case__	       =   output.images
		assert image.shape[0] == 2

		snake_case__	       =   torch.Generator(device=UpperCamelCase_		).manual_seed(0		)
		snake_case__	       =   sd_pipe(
		    [prompt]      ,       image=UpperCamelCase_      ,       generator=UpperCamelCase_      ,       num_images_per_prompt=2      ,       guidance_scale=6.0      ,       noise_level=20      ,       num_inference_steps=2      ,       output_type='np'      ,       )
		snake_case__	       =   output.images
		assert image.shape[0] == 2





	@unittest.skipIf(torch_device != 'cuda'      ,       'This test requires a GPU'		)
	def   _snake_case   (		self		)      ->		str:
		snake_case__	       =   self.dummy_cond_unet_upscale
		snake_case__	       =   DDPMScheduler()
		snake_case__	       =   DDIMScheduler(prediction_type='v_prediction'		)
		snake_case__	       =   self.dummy_vae
		snake_case__	       =   self.dummy_text_encoder
		snake_case__	       =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

		snake_case__	       =   self.dummy_image.cpu().permute(0      ,       2      ,       3      ,       1		)[0]
		snake_case__	       =   Image.fromarray(np.uinta(UpperCamelCase_		)		).convert('RGB'		).resize((64, 64)		)

		# put models in fp16, except vae as it overflows in fp16
		snake_case__	       =   unet.half()
		snake_case__	       =   text_encoder.half()

		# make sure here that pndm scheduler skips prk
		snake_case__	       =   StableDiffusionUpscalePipeline(
		    unet=UpperCamelCase_      ,       low_res_scheduler=UpperCamelCase_      ,       scheduler=UpperCamelCase_      ,       vae=UpperCamelCase_      ,       text_encoder=UpperCamelCase_      ,       tokenizer=UpperCamelCase_      ,       max_noise_level=350      ,       )
		snake_case__	       =   sd_pipe.to(UpperCamelCase_		)
		sd_pipe.set_progress_bar_config(disable=UpperCamelCase_		)

		snake_case__	       =   'A painting of a squirrel eating a burger'
		snake_case__	       =   torch.manual_seed(0		)
		snake_case__	       =   sd_pipe(
		    [prompt]      ,       image=UpperCamelCase_      ,       generator=UpperCamelCase_      ,       num_inference_steps=2      ,       output_type='np'      ,       ).images

		snake_case__	       =   low_res_image.size[0] * 4
		assert image.shape == (1, expected_height_width, expected_height_width, 3)






@slow
@require_torch_gpu
class 				__snake_case     (					unittest.TestCase							):





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





	def   _snake_case   (		self		)      ->		Optional[int]:
		snake_case__	       =   load_image(
		    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
		    '/sd2-upscale/low_res_cat.png'		)
		snake_case__	       =   load_numpy(
		    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale'
		    '/upsampled_cat.npy'		)

		snake_case__	       =   'stabilityai/stable-diffusion-x4-upscaler'
		snake_case__	       =   StableDiffusionUpscalePipeline.from_pretrained(UpperCamelCase_		)
		pipe.to(UpperCamelCase_		)
		pipe.set_progress_bar_config(disable=UpperCamelCase_		)
		pipe.enable_attention_slicing()

		snake_case__	       =   'a cat sitting on a park bench'

		snake_case__	       =   torch.manual_seed(0		)
		snake_case__	       =   pipe(
		    prompt=UpperCamelCase_      ,       image=UpperCamelCase_      ,       generator=UpperCamelCase_      ,       output_type='np'      ,       )
		snake_case__	       =   output.images[0]

		assert image.shape == (512, 512, 3)
		assert np.abs(expected_image - image		).max() < 1E-3





	def   _snake_case   (		self		)      ->		List[Any]:
		snake_case__	       =   load_image(
		    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
		    '/sd2-upscale/low_res_cat.png'		)
		snake_case__	       =   load_numpy(
		    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale'
		    '/upsampled_cat_fp16.npy'		)

		snake_case__	       =   'stabilityai/stable-diffusion-x4-upscaler'
		snake_case__	       =   StableDiffusionUpscalePipeline.from_pretrained(
		    UpperCamelCase_      ,       torch_dtype=torch.floataa      ,       )
		pipe.to(UpperCamelCase_		)
		pipe.set_progress_bar_config(disable=UpperCamelCase_		)
		pipe.enable_attention_slicing()

		snake_case__	       =   'a cat sitting on a park bench'

		snake_case__	       =   torch.manual_seed(0		)
		snake_case__	       =   pipe(
		    prompt=UpperCamelCase_      ,       image=UpperCamelCase_      ,       generator=UpperCamelCase_      ,       output_type='np'      ,       )
		snake_case__	       =   output.images[0]

		assert image.shape == (512, 512, 3)
		assert np.abs(expected_image - image		).max() < 5E-1





	def   _snake_case   (		self		)      ->		int:
		torch.cuda.empty_cache()
		torch.cuda.reset_max_memory_allocated()
		torch.cuda.reset_peak_memory_stats()

		snake_case__	       =   load_image(
		    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
		    '/sd2-upscale/low_res_cat.png'		)

		snake_case__	       =   'stabilityai/stable-diffusion-x4-upscaler'
		snake_case__	       =   StableDiffusionUpscalePipeline.from_pretrained(
		    UpperCamelCase_      ,       torch_dtype=torch.floataa      ,       )
		pipe.to(UpperCamelCase_		)
		pipe.set_progress_bar_config(disable=UpperCamelCase_		)
		pipe.enable_attention_slicing(1		)
		pipe.enable_sequential_cpu_offload()

		snake_case__	       =   'a cat sitting on a park bench'

		snake_case__	       =   torch.manual_seed(0		)
		snake_case__	       =   pipe(
		    prompt=UpperCamelCase_      ,       image=UpperCamelCase_      ,       generator=UpperCamelCase_      ,       num_inference_steps=5      ,       output_type='np'      ,       )

		snake_case__	       =   torch.cuda.max_memory_allocated()
		# make sure that less than 2.9 GB is allocated
		assert mem_bytes < 2.9 * 10**9



