




'''simple docstring'''

import tempfile
import unittest

import numpy as np
from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import BertConfig, is_flax_available
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_flax


if is_flax_available():
							import os

							from flax.core.frozen_dict import unfreeze
							from flax.traverse_util import flatten_dict

							from transformers import FlaxBertModel

							a  :				List[str]									=  '0.12'  # assumed parallelism: 8






@require_flax
@is_staging_test
class 						a				(   unittest.TestCase     ):


				@classmethod
				def   A_ (		cls    :					str					):
									snake_case_									=				TOKEN
									HfFolder.save_token(lowercase_					)


				@classmethod
				def   A_ (		cls    :					Optional[Any]					):
									try:
														delete_repo(token=cls._token       ,				repo_id='''test-model-flax'''					)
									except HTTPError:
														pass

									try:
														delete_repo(token=cls._token       ,				repo_id='''valid_org/test-model-flax-org'''					)
									except HTTPError:
														pass


				def   A_ (		self    :					Tuple					):
									snake_case_									=				BertConfig(
									    vocab_size=99       ,				hidden_size=32       ,				num_hidden_layers=5       ,				num_attention_heads=4       ,				intermediate_size=37					)
									snake_case_									=				FlaxBertModel(lowercase_					)
									model.push_to_hub('''test-model-flax'''       ,				use_auth_token=self._token					)

									snake_case_									=				FlaxBertModel.from_pretrained(F"{USER}/test-model-flax"					)

									snake_case_									=				flatten_dict(unfreeze(model.params					)					)
									snake_case_									=				flatten_dict(unfreeze(new_model.params					)					)

									for key in base_params.keys():
														snake_case_									=				(base_params[key] - new_params[key]).sum().item()
														self.assertLessEqual(lowercase_       ,				1e-3       ,				msg=F"{key} not identical"					)

									# Reset repo
									delete_repo(token=self._token       ,				repo_id='''test-model-flax'''					)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
														model.save_pretrained(lowercase_       ,				repo_id='''test-model-flax'''       ,				push_to_hub=lowercase_       ,				use_auth_token=self._token					)

									snake_case_									=				FlaxBertModel.from_pretrained(F"{USER}/test-model-flax"					)

									snake_case_									=				flatten_dict(unfreeze(model.params					)					)
									snake_case_									=				flatten_dict(unfreeze(new_model.params					)					)

									for key in base_params.keys():
														snake_case_									=				(base_params[key] - new_params[key]).sum().item()
														self.assertLessEqual(lowercase_       ,				1e-3       ,				msg=F"{key} not identical"					)


				def   A_ (		self    :					Optional[Any]					):
									snake_case_									=				BertConfig(
									    vocab_size=99       ,				hidden_size=32       ,				num_hidden_layers=5       ,				num_attention_heads=4       ,				intermediate_size=37					)
									snake_case_									=				FlaxBertModel(lowercase_					)
									model.push_to_hub('''valid_org/test-model-flax-org'''       ,				use_auth_token=self._token					)

									snake_case_									=				FlaxBertModel.from_pretrained('''valid_org/test-model-flax-org'''					)

									snake_case_									=				flatten_dict(unfreeze(model.params					)					)
									snake_case_									=				flatten_dict(unfreeze(new_model.params					)					)

									for key in base_params.keys():
														snake_case_									=				(base_params[key] - new_params[key]).sum().item()
														self.assertLessEqual(lowercase_       ,				1e-3       ,				msg=F"{key} not identical"					)

									# Reset repo
									delete_repo(token=self._token       ,				repo_id='''valid_org/test-model-flax-org'''					)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
														model.save_pretrained(
														    lowercase_       ,				repo_id='''valid_org/test-model-flax-org'''       ,				push_to_hub=lowercase_       ,				use_auth_token=self._token					)

									snake_case_									=				FlaxBertModel.from_pretrained('''valid_org/test-model-flax-org'''					)

									snake_case_									=				flatten_dict(unfreeze(model.params					)					)
									snake_case_									=				flatten_dict(unfreeze(new_model.params					)					)

									for key in base_params.keys():
														snake_case_									=				(base_params[key] - new_params[key]).sum().item()
														self.assertLessEqual(lowercase_       ,				1e-3       ,				msg=F"{key} not identical"					)







def 		__magic_name__    (	__UpperCAmelCase,      __UpperCAmelCase	)		->					Union[str, Any]:

					'''simple docstring'''

					snake_case_									=				True
					snake_case_									=				flatten_dict(modela.params	)
					snake_case_									=				flatten_dict(modela.params	)
					for key in flat_params_a.keys():
										if np.sum(np.abs(flat_params_a[key] - flat_params_a[key]	)	) > 1e-4:
															snake_case_									=				False

					return models_are_equal






@require_flax
class 						a				(   unittest.TestCase     ):


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				BertConfig.from_pretrained('''hf-internal-testing/tiny-bert-flax-only'''					)
									snake_case_									=				FlaxBertModel(lowercase_					)

									snake_case_									=				'''bert'''
									with tempfile.TemporaryDirectory() as tmp_dir:
														model.save_pretrained(os.path.join(lowercase_       ,				lowercase_					)					)

														with self.assertRaises(lowercase_					):
																			snake_case_									=				FlaxBertModel.from_pretrained(lowercase_					)

														snake_case_									=				FlaxBertModel.from_pretrained(lowercase_       ,				subfolder=lowercase_					)

									self.assertTrue(check_models_equal(lowercase_       ,				lowercase_					)					)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				BertConfig.from_pretrained('''hf-internal-testing/tiny-bert-flax-only'''					)
									snake_case_									=				FlaxBertModel(lowercase_					)

									snake_case_									=				'''bert'''
									with tempfile.TemporaryDirectory() as tmp_dir:
														model.save_pretrained(os.path.join(lowercase_       ,				lowercase_					)       ,				max_shard_size='''10KB'''					)

														with self.assertRaises(lowercase_					):
																			snake_case_									=				FlaxBertModel.from_pretrained(lowercase_					)

														snake_case_									=				FlaxBertModel.from_pretrained(lowercase_       ,				subfolder=lowercase_					)

									self.assertTrue(check_models_equal(lowercase_       ,				lowercase_					)					)


				def   A_ (		self    :					str					):
									snake_case_									=				'''bert'''
									snake_case_									=				'''hf-internal-testing/tiny-random-bert-subfolder'''

									with self.assertRaises(lowercase_					):
														snake_case_									=				FlaxBertModel.from_pretrained(lowercase_					)

									snake_case_									=				FlaxBertModel.from_pretrained(lowercase_       ,				subfolder=lowercase_					)

									self.assertIsNotNone(lowercase_					)


				def   A_ (		self    :					Tuple					):
									snake_case_									=				'''bert'''
									snake_case_									=				'''hf-internal-testing/tiny-random-bert-sharded-subfolder'''
									with self.assertRaises(lowercase_					):
														snake_case_									=				FlaxBertModel.from_pretrained(lowercase_					)

									snake_case_									=				FlaxBertModel.from_pretrained(lowercase_       ,				subfolder=lowercase_					)

									self.assertIsNotNone(lowercase_					)







'''simple docstring'''

import gc
import unittest

import torch
from parameterized import parameterized

from diffusers import AutoencoderKL
from diffusers.utils import floats_tensor, load_hf_numpy, require_torch_gpu, slow, torch_all_close, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()






class 						a				(   _lowerCamelCase   ,     _lowerCamelCase   ,     unittest.TestCase     ):
				snake_case_      =		AutoencoderKL
				snake_case_      =		"sample"
				snake_case_      =		1e-2


				@property
				def   A_ (		self    :					Dict					):
									snake_case_									=				4
									snake_case_									=				3
									snake_case_									=				(32, 32)

									snake_case_									=				floats_tensor((batch_size, num_channels) + sizes					).to(lowercase_					)

									return {"sample": image}


				@property
				def   A_ (		self    :					List[Any]					):
									return (3, 32, 32)


				@property
				def   A_ (		self    :					Dict					):
									return (3, 32, 32)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				{
									    '''block_out_channels''': [32, 64],
									    '''in_channels''': 3,
									    '''out_channels''': 3,
									    '''down_block_types''': ['''DownEncoderBlock2D''', '''DownEncoderBlock2D'''],
									    '''up_block_types''': ['''UpDecoderBlock2D''', '''UpDecoderBlock2D'''],
									    '''latent_channels''': 4,
									}
									snake_case_									=				self.dummy_input
									return init_dict, inputs_dict


				def   A_ (		self    :					Any					):
									pass


				def   A_ (		self    :					str					):
									pass


				@unittest.skipIf(torch_device == '''mps'''       ,				'''Gradient checkpointing skipped on MPS'''					)
				def   A_ (		self    :					Dict					):
									# enable deterministic behavior for gradient checkpointing
									snake_case_			,snake_case_									=				self.prepare_init_args_and_inputs_for_common()
									snake_case_									=				self.model_class(**lowercase_					)
									model.to(lowercase_					)

									assert not model.is_gradient_checkpointing and model.training

									snake_case_									=				model(**lowercase_					).sample
									# run the backwards pass on the model. For backwards pass, for simplicity purpose,
									# we won't calculate the loss and rather backprop on out.sum()
									model.zero_grad()

									snake_case_									=				torch.randn_like(lowercase_					)
									snake_case_									=				(out - labels).mean()
									loss.backward()

									# re-instantiate the model now enabling gradient checkpointing
									snake_case_									=				self.model_class(**lowercase_					)
									# clone model
									model_a.load_state_dict(model.state_dict()					)
									model_a.to(lowercase_					)
									model_a.enable_gradient_checkpointing()

									assert model_a.is_gradient_checkpointing and model_a.training

									snake_case_									=				model_a(**lowercase_					).sample
									# run the backwards pass on the model. For backwards pass, for simplicity purpose,
									# we won't calculate the loss and rather backprop on out.sum()
									model_a.zero_grad()
									snake_case_									=				(out_a - labels).mean()
									loss_a.backward()

									# compare the output and parameters gradients
									self.assertTrue((loss - loss_a).abs() < 1e-5					)
									snake_case_									=				dict(model.named_parameters()					)
									snake_case_									=				dict(model_a.named_parameters()					)
									for name, param in named_params.items():
														self.assertTrue(torch_all_close(param.grad.data       ,				named_params_a[name].grad.data       ,				atol=5e-5					)					)


				def   A_ (		self    :					Tuple					):
									snake_case_			,snake_case_									=				AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''       ,				output_loading_info=lowercase_					)
									self.assertIsNotNone(lowercase_					)
									self.assertEqual(len(loading_info['''missing_keys''']					)       ,				0					)

									model.to(lowercase_					)
									snake_case_									=				model(**self.dummy_input					)

									assert image is not None, "Make sure output is not None"


				def   A_ (		self    :					Tuple					):
									snake_case_									=				AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''					)
									snake_case_									=				model.to(lowercase_					)
									model.eval()

									if torch_device == "mps":
														snake_case_									=				torch.manual_seed(0					)
									else:
														snake_case_									=				torch.Generator(device=lowercase_					).manual_seed(0					)

									snake_case_									=				torch.randn(
									    1       ,				model.config.in_channels       ,				model.config.sample_size       ,				model.config.sample_size       ,				generator=torch.manual_seed(0					)       ,				)
									snake_case_									=				image.to(lowercase_					)
									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				sample_posterior=lowercase_       ,				generator=lowercase_					).sample

									snake_case_									=				output[0, -1, -3:, -3:].flatten().cpu()

									# Since the VAE Gaussian prior's generator is seeded on the appropriate device,
									# the expected output slices are not the same for CPU and GPU.
									if torch_device == "mps":
														snake_case_									=				torch.tensor(
														    [
														        -4.0_078e-01,
														        -3.8_323e-04,
														        -1.2_681e-01,
														        -1.1_462e-01,
														        2.0_095e-01,
														        1.0_893e-01,
														        -8.8_247e-02,
														        -3.0_361e-01,
														        -9.8_644e-03,
														    ]					)
									elif torch_device == "cpu":
														snake_case_									=				torch.tensor(
														    [-0.1352, 0.0878, 0.0419, -0.0818, -0.1069, 0.0688, -0.1458, -0.4446, -0.0026]					)
									else:
														snake_case_									=				torch.tensor(
														    [-0.2421, 0.4642, 0.2507, -0.0438, 0.0682, 0.3160, -0.2018, -0.0727, 0.2485]					)

									self.assertTrue(torch_all_close(lowercase_       ,				lowercase_       ,				rtol=1e-2					)					)







@slow
class 						a				(   unittest.TestCase     ):


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Optional[int]					):
									return F"gaussian_noise_s={seed}_shape={'_'.join([str(lowercase_					) for s in shape]					)}.npy"


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


				def   A_ (		self    :					Dict       ,				lowercase_    :					List[Any]=0       ,				lowercase_    :					Union[str, Any]=(4, 3, 512, 512)       ,				lowercase_    :					Optional[Any]=False					):
									snake_case_									=				torch.floataa if fpaa else torch.floataa
									snake_case_									=				torch.from_numpy(load_hf_numpy(self.get_file_format(lowercase_       ,				lowercase_					)					)					).to(lowercase_					).to(lowercase_					)
									return image


				def   A_ (		self    :					Any       ,				lowercase_    :					Dict="CompVis/stable-diffusion-v1-4"       ,				lowercase_    :					List[str]=False					):
									snake_case_									=				'''fp16''' if fpaa else None
									snake_case_									=				torch.floataa if fpaa else torch.floataa

									snake_case_									=				AutoencoderKL.from_pretrained(
									    lowercase_       ,				subfolder='''vae'''       ,				torch_dtype=lowercase_       ,				revision=lowercase_       ,				)
									model.to(lowercase_					).eval()

									return model


				def   A_ (		self    :					Any       ,				lowercase_    :					int=0					):
									if torch_device == "mps":
														return torch.manual_seed(lowercase_					)
									return torch.Generator(device=lowercase_					).manual_seed(lowercase_					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1603, 0.9878, -0.0495, -0.0790, -0.2709, 0.8375, -0.2060, -0.0824], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2376, 0.1168, 0.1332, -0.4840, -0.2508, -0.0791, -0.0493, -0.4089], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				generator=lowercase_       ,				sample_posterior=lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, -2:, :2].flatten().float().cpu()
									snake_case_									=				torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=3e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.0513, 0.0289, 1.3799, 0.2166, -0.2573, -0.0871, 0.5103, -0.0999]],
				        [47, [-0.4128, -0.1320, -0.3704, 0.1965, -0.4116, -0.2332, -0.3340, 0.2247]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Optional[int]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					Dict					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				fpaa=lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				generator=lowercase_       ,				sample_posterior=lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().float().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-2					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1609, 0.9866, -0.0487, -0.0777, -0.2716, 0.8368, -0.2055, -0.0814], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2377, 0.1147, 0.1333, -0.4841, -0.2506, -0.0805, -0.0491, -0.4085], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					Tuple       ,				lowercase_    :					Dict       ,				lowercase_    :					str       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, -2:, :2].flatten().float().cpu()
									snake_case_									=				torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=3e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [13, [-0.2051, -0.1803, -0.2311, -0.2114, -0.3292, -0.3574, -0.2953, -0.3323]],
				        [37, [-0.2632, -0.2625, -0.2199, -0.2741, -0.4539, -0.4990, -0.3720, -0.4925]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Dict       ,				lowercase_    :					Tuple       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [27, [-0.0369, 0.0207, -0.0776, -0.0682, -0.1747, -0.1930, -0.1465, -0.2039]],
				        [16, [-0.1628, -0.2134, -0.2747, -0.2642, -0.3774, -0.4404, -0.3687, -0.4277]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Any       ,				lowercase_    :					Optional[Any]					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)       ,				fpaa=lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().float().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=5e-3					)


				@parameterized.expand([(13,), (16,), (27,)]					)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()       ,				reason='''xformers is not required when using PyTorch 2.0.'''					)
				def   A_ (		self    :					Optional[Any]       ,				lowercase_    :					List[str]					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)       ,				fpaa=lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									model.enable_xformers_memory_efficient_attention()
									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-1					)


				@parameterized.expand([(13,), (16,), (37,)]					)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()       ,				reason='''xformers is not required when using PyTorch 2.0.'''					)
				def   A_ (		self    :					Optional[Any]       ,				lowercase_    :					Any					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									model.enable_xformers_memory_efficient_attention()
									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-2					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.3001, 0.0918, -2.6984, -3.9720, -3.2099, -5.0353, 1.7338, -0.2065, 3.4267]],
				        [47, [-1.5030, -4.3871, -6.0355, -9.1157, -1.6661, -2.7853, 2.1607, -5.0823, 2.5633]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					str       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					Tuple					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.encode(lowercase_					).latent_dist
														snake_case_									=				dist.sample(generator=lowercase_					)

									assert list(sample.shape					) == [image.shape[0], 4] + [i // 8 for i in image.shape[2:]]

									snake_case_									=				sample[0, -1, -3:, -3:].flatten().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									snake_case_									=				3e-3 if torch_device != '''mps''' else 1e-2
									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=lowercase_					)

