


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




							"""simple docstring"""


							__UpperCAmelCase   :							str					  =  AutoencoderKL
							__UpperCAmelCase   :							str					  =  """sample"""
							__UpperCAmelCase   :							List[str]					  =  1e-2

							@property
							def    __snake_case (							self	:  Optional[int]    )     ->   Optional[Any]:
														__snake_case		:      Optional[Any]										=       4
														__snake_case		:      Any										=       3
														__snake_case		:      Optional[int]										=       (32, 32)

														__snake_case		:      List[str]										=       floats_tensor((batch_size, num_channels) + sizes    ).to(_A    )

														return {"sample": image}

							@property
							def    __snake_case (							self	:  int    )     ->   Dict:
														return (3, 32, 32)

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

							def    __snake_case (							self	:  Any    )     ->   str:
														__snake_case		:      Optional[int]										=       {
														    '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		:      int										=       self.dummy_input
														return init_dict, inputs_dict

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

							def    __snake_case (							self	:  Tuple    )     ->   Any:
														pass

							@unittest.skipIf(torch_device == "mps"   ,     "Gradient checkpointing skipped on MPS"    )
							def    __snake_case (							self	:  Optional[Any]    )     ->   List[Any]:
														# enable deterministic behavior for gradient checkpointing
														__snake_case		:      List[Any]										=       self.prepare_init_args_and_inputs_for_common()
														__snake_case		:      int										=       self.model_class(**_A    )
														model.to(_A    )

														assert not model.is_gradient_checkpointing and model.training

														__snake_case		:      List[Any]										=       model(**_A    ).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		:      str										=       torch.randn_like(_A    )
														__snake_case		:      List[Any]										=       (out - labels).mean()
														loss.backward()

														# re-instantiate the model now enabling gradient checkpointing
														__snake_case		:      Optional[Any]										=       self.model_class(**_A    )
														# clone model
														model_a.load_state_dict(model.state_dict()    )
														model_a.to(_A    )
														model_a.enable_gradient_checkpointing()

														assert model_a.is_gradient_checkpointing and model_a.training

														__snake_case		:      List[Any]										=       model_a(**_A    ).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		:      Dict										=       (out_a - labels).mean()
														loss_a.backward()

														# compare the output and parameters gradients
														self.assertTrue((loss - loss_a).abs() < 1E-5    )
														__snake_case		:      List[str]										=       dict(model.named_parameters()    )
														__snake_case		:      Union[str, Any]										=       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    __snake_case (							self	:  Union[str, Any]    )     ->   Any:
														__snake_case		:      List[Any]										=       AutoencoderKL.from_pretrained("fusing/autoencoder-kl-dummy"   ,     output_loading_info=_A    )
														self.assertIsNotNone(_A    )
														self.assertEqual(len(loading_info["missing_keys"]    )   ,     0    )

														model.to(_A    )
														__snake_case		:      str										=       model(**self.dummy_input    )

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






							def    __snake_case (							self	:  Optional[Any]    )     ->   str:
														__snake_case		:      Optional[int]										=       AutoencoderKL.from_pretrained("fusing/autoencoder-kl-dummy"    )
														__snake_case		:      List[Any]										=       model.to(_A    )
														model.eval()

														if torch_device == "mps":
																					__snake_case		:      str										=       torch.manual_seed(0    )
														else:
																					__snake_case		:      str										=       torch.Generator(device=_A    ).manual_seed(0    )

														__snake_case		:      Optional[int]										=       torch.randn(
														    1   ,     model.config.in_channels   ,     model.config.sample_size   ,     model.config.sample_size   ,     generator=torch.manual_seed(0    )   ,     )
														__snake_case		:      List[Any]										=       image.to(_A    )
														with torch.no_grad():
																					__snake_case		:      str										=       model(_A   ,     sample_posterior=_A   ,     generator=_A    ).sample

														__snake_case		:      Tuple										=       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		:      int										=       torch.tensor(
																					    [
																					        -4.0078E-01,
																					        -3.8323E-04,
																					        -1.2681E-01,
																					        -1.1462E-01,
																					        2.0095E-01,
																					        1.0893E-01,
																					        -8.8247E-02,
																					        -3.0361E-01,
																					        -9.8644E-03,
																					    ]    )
														elif torch_device == "cpu":
																					__snake_case		:      Any										=       torch.tensor(
																					    [-0.13_52, 0.08_78, 0.04_19, -0.08_18, -0.10_69, 0.06_88, -0.14_58, -0.44_46, -0.00_26]    )
														else:
																					__snake_case		:      int										=       torch.tensor(
																					    [-0.24_21, 0.46_42, 0.25_07, -0.04_38, 0.06_82, 0.31_60, -0.20_18, -0.07_27, 0.24_85]    )

														self.assertTrue(torch_all_close(_A   ,     _A   ,     rtol=1E-2    )    )

@slow
class     a		(unittest.TestCase				):




							"""simple docstring"""


							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  Tuple    )     ->   List[Any]:
														return F'gaussian_noise_s={seed}_shape={"_".join([str(_A    ) for s in shape]    )}.npy'

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

							def    __snake_case (							self	:  str   ,     lowerCamelCase	:  Optional[Any]=0   ,     lowerCamelCase	:  List[str]=(4, 3, 512, 512)   ,     lowerCamelCase	:  Union[str, Any]=False    )     ->   int:
														__snake_case		:      Optional[int]										=       torch.floataa if fpaa else torch.floataa
														__snake_case		:      Optional[Any]										=       torch.from_numpy(load_hf_numpy(self.get_file_format(_A   ,     _A    )    )    ).to(_A    ).to(_A    )
														return image

							def    __snake_case (							self	:  Dict   ,     lowerCamelCase	:  int="CompVis/stable-diffusion-v1-4"   ,     lowerCamelCase	:  Any=False    )     ->   List[Any]:
														__snake_case		:      Dict										=       'fp16' if fpaa else None
														__snake_case		:      Any										=       torch.floataa if fpaa else torch.floataa

														__snake_case		:      Union[str, Any]										=       AutoencoderKL.from_pretrained(
														    _A   ,     subfolder="vae"   ,     torch_dtype=_A   ,     revision=_A   ,     )
														model.to(_A    ).eval()

														return model

							def    __snake_case (							self	:  Union[str, Any]   ,     lowerCamelCase	:  Any=0    )     ->   Union[str, Any]:
														if torch_device == "mps":
																					return torch.manual_seed(_A    )
														return torch.Generator(device=_A    ).manual_seed(_A    )

							@parameterized.expand(
							    [
							        # fmt: off
							        [33, [-0.16_03, 0.98_78, -0.04_95, -0.07_90, -0.27_09, 0.83_75, -0.20_60, -0.08_24], [-0.23_95, 0.00_98, 0.01_02, -0.07_09, -0.28_40, -0.02_74, -0.07_18, -0.18_24]],
							        [47, [-0.23_76, 0.11_68, 0.13_32, -0.48_40, -0.25_08, -0.07_91, -0.04_93, -0.40_89], [0.03_50, 0.08_47, 0.04_67, 0.03_44, -0.08_42, -0.05_47, -0.06_33, -0.11_31]],
							        # fmt: on
							    ]    )
							def    __snake_case (							self	:  Union[str, Any]   ,     lowerCamelCase	:  Optional[int]   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  Optional[Any]    )     ->   int:
														__snake_case		:      List[Any]										=       self.get_sd_vae_model()
														__snake_case		:      Optional[int]										=       self.get_sd_image(_A    )
														__snake_case		:      Any										=       self.get_generator(_A    )

														with torch.no_grad():
																					__snake_case		:      Optional[Any]										=       model(_A   ,     generator=_A   ,     sample_posterior=_A    ).sample

														assert sample.shape == image.shape

														__snake_case		:      Dict										=       sample[-1, -2:, -2:, :2].flatten().float().cpu()
														__snake_case		:      Tuple										=       torch.tensor(expected_slice_mps if torch_device == "mps" else expected_slice    )

														assert torch_all_close(_A   ,     _A   ,     atol=3E-3    )

							@parameterized.expand(
							    [
							        # fmt: off
							        [33, [-0.05_13, 0.02_89, 1.37_99, 0.21_66, -0.25_73, -0.08_71, 0.51_03, -0.09_99]],
							        [47, [-0.41_28, -0.13_20, -0.37_04, 0.19_65, -0.41_16, -0.23_32, -0.33_40, 0.22_47]],
							        # fmt: on
							    ]    )
							@require_torch_gpu
							def    __snake_case (							self	:  Union[str, Any]   ,     lowerCamelCase	:  Tuple   ,     lowerCamelCase	:  str    )     ->   str:
														__snake_case		:      int										=       self.get_sd_vae_model(fpaa=_A    )
														__snake_case		:      List[Any]										=       self.get_sd_image(_A   ,     fpaa=_A    )
														__snake_case		:      Any										=       self.get_generator(_A    )

														with torch.no_grad():
																					__snake_case		:      Optional[int]										=       model(_A   ,     generator=_A   ,     sample_posterior=_A    ).sample

														assert sample.shape == image.shape

														__snake_case		:      int										=       sample[-1, -2:, :2, -2:].flatten().float().cpu()
														__snake_case		:      Any										=       torch.tensor(_A    )

														assert torch_all_close(_A   ,     _A   ,     atol=1E-2    )

							@parameterized.expand(
							    [
							        # fmt: off
							        [33, [-0.16_09, 0.98_66, -0.04_87, -0.07_77, -0.27_16, 0.83_68, -0.20_55, -0.08_14], [-0.23_95, 0.00_98, 0.01_02, -0.07_09, -0.28_40, -0.02_74, -0.07_18, -0.18_24]],
							        [47, [-0.23_77, 0.11_47, 0.13_33, -0.48_41, -0.25_06, -0.08_05, -0.04_91, -0.40_85], [0.03_50, 0.08_47, 0.04_67, 0.03_44, -0.08_42, -0.05_47, -0.06_33, -0.11_31]],
							        # fmt: on
							    ]    )
							def    __snake_case (							self	:  Optional[Any]   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  Dict    )     ->   Any:
														__snake_case		:      Any										=       self.get_sd_vae_model()
														__snake_case		:      Dict										=       self.get_sd_image(_A    )

														with torch.no_grad():
																					__snake_case		:      Tuple										=       model(_A    ).sample

														assert sample.shape == image.shape

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

														assert torch_all_close(_A   ,     _A   ,     atol=3E-3    )

							@parameterized.expand(
							    [
							        # fmt: off
							        [13, [-0.20_51, -0.18_03, -0.23_11, -0.21_14, -0.32_92, -0.35_74, -0.29_53, -0.33_23]],
							        [37, [-0.26_32, -0.26_25, -0.21_99, -0.27_41, -0.45_39, -0.49_90, -0.37_20, -0.49_25]],
							        # fmt: on
							    ]    )
							@require_torch_gpu
							def    __snake_case (							self	:  Tuple   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Tuple    )     ->   str:
														__snake_case		:      Tuple										=       self.get_sd_vae_model()
														__snake_case		:      List[Any]										=       self.get_sd_image(_A   ,     shape=(3, 4, 64, 64)    )

														with torch.no_grad():
																					__snake_case		:      Tuple										=       model.decode(_A    ).sample

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

														__snake_case		:      Any										=       sample[-1, -2:, :2, -2:].flatten().cpu()
														__snake_case		:      List[str]										=       torch.tensor(_A    )

														assert torch_all_close(_A   ,     _A   ,     atol=1E-3    )

							@parameterized.expand(
							    [
							        # fmt: off
							        [27, [-0.03_69, 0.02_07, -0.07_76, -0.06_82, -0.17_47, -0.19_30, -0.14_65, -0.20_39]],
							        [16, [-0.16_28, -0.21_34, -0.27_47, -0.26_42, -0.37_74, -0.44_04, -0.36_87, -0.42_77]],
							        # fmt: on
							    ]    )
							@require_torch_gpu
							def    __snake_case (							self	:  Dict   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  Dict    )     ->   Optional[int]:
														__snake_case		:      Optional[int]										=       self.get_sd_vae_model(fpaa=_A    )
														__snake_case		:      List[str]										=       self.get_sd_image(_A   ,     shape=(3, 4, 64, 64)   ,     fpaa=_A    )

														with torch.no_grad():
																					__snake_case		:      Tuple										=       model.decode(_A    ).sample

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

														__snake_case		:      Optional[Any]										=       sample[-1, -2:, :2, -2:].flatten().float().cpu()
														__snake_case		:      Optional[Any]										=       torch.tensor(_A    )

														assert torch_all_close(_A   ,     _A   ,     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    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  Union[str, Any]    )     ->   List[str]:
														__snake_case		:      Any										=       self.get_sd_vae_model(fpaa=_A    )
														__snake_case		:      int										=       self.get_sd_image(_A   ,     shape=(3, 4, 64, 64)   ,     fpaa=_A    )

														with torch.no_grad():
																					__snake_case		:      Optional[Any]										=       model.decode(_A    ).sample

														model.enable_xformers_memory_efficient_attention()
														with torch.no_grad():
																					__snake_case		:      Tuple										=       model.decode(_A    ).sample

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

														assert torch_all_close(_A   ,     _A   ,     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    __snake_case (							self	:  Dict   ,     lowerCamelCase	:  Optional[Any]    )     ->   Tuple:
														__snake_case		:      Union[str, Any]										=       self.get_sd_vae_model()
														__snake_case		:      Tuple										=       self.get_sd_image(_A   ,     shape=(3, 4, 64, 64)    )

														with torch.no_grad():
																					__snake_case		:      List[Any]										=       model.decode(_A    ).sample

														model.enable_xformers_memory_efficient_attention()
														with torch.no_grad():
																					__snake_case		:      Any										=       model.decode(_A    ).sample

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

														assert torch_all_close(_A   ,     _A   ,     atol=1E-2    )






							@parameterized.expand(
							    [
							        # fmt: off
							        [33, [-0.30_01, 0.09_18, -2.69_84, -3.97_20, -3.20_99, -5.03_53, 1.73_38, -0.20_65, 3.42_67]],
							        [47, [-1.50_30, -4.38_71, -6.03_55, -9.11_57, -1.66_61, -2.78_53, 2.16_07, -5.08_23, 2.56_33]],
							        # fmt: on
							    ]    )
							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  Optional[Any]    )     ->   Dict:
														__snake_case		:      Optional[Any]										=       self.get_sd_vae_model()
														__snake_case		:      Union[str, Any]										=       self.get_sd_image(_A    )
														__snake_case		:      Any										=       self.get_generator(_A    )

														with torch.no_grad():
																					__snake_case		:      Dict										=       model.encode(_A    ).latent_dist
																					__snake_case		:      Dict										=       dist.sample(generator=_A    )

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

														__snake_case		:      str										=       sample[0, -1, -3:, -3:].flatten().cpu()
														__snake_case		:      List[str]										=       torch.tensor(_A    )

														__snake_case		:      Any										=       3E-3 if torch_device != 'mps' else 1E-2
														assert torch_all_close(_A   ,     _A   ,     atol=_A    )





from typing import Dict

import numpy as np
import torch

from . import residue_constants as rc
from .tensor_utils import tensor_tree_map, tree_map







def     lowerCAmelCase_					( __lowerCamelCase     ):
							__snake_case		:      Any										=       []
							__snake_case		:      Optional[Any]										=       []
							__snake_case		:      List[Any]										=       []

							for rt in rc.restypes:
														__snake_case		:      Any										=       rc.restype_name_to_atomaa_names[rc.restype_atoa[rt]]
														restype_atomaa_to_atomaa_list.append([(rc.atom_order[name] if name else 0) for name in atom_names]     )
														__snake_case		:      Tuple										=       {name: i for i, name in enumerate(__lowerCamelCase     )}
														restype_atomaa_to_atomaa_list.append(
														    [(atom_name_to_idxaa[name] if name in atom_name_to_idxaa else 0) for name in rc.atom_types]     )

														restype_atomaa_mask_list.append([(1.0 if name else 0.0) for name in atom_names]     )

							# Add dummy mapping for restype 'UNK'
							restype_atomaa_to_atomaa_list.append([0] * 1_4     )
							restype_atomaa_to_atomaa_list.append([0] * 3_7     )
							restype_atomaa_mask_list.append([0.0] * 1_4     )

							__snake_case		:      int										=       torch.tensor(
							    __lowerCamelCase							,     dtype=torch.intaa							,     device=protein["aatype"].device							,     )
							__snake_case		:      List[str]										=       torch.tensor(
							    __lowerCamelCase							,     dtype=torch.intaa							,     device=protein["aatype"].device							,     )
							__snake_case		:      Optional[Any]										=       torch.tensor(
							    __lowerCamelCase							,     dtype=torch.floataa							,     device=protein["aatype"].device							,     )
							__snake_case		:      Optional[int]										=       protein["aatype"].to(torch.long     )

							# create the mapping for (residx, atom14) --> atom37, i.e. an array
							# with shape (num_res, 14) containing the atom37 indices for this protein
							__snake_case		:      Optional[Any]										=       restype_atomaa_to_atomaa[protein_aatype]
							__snake_case		:      Tuple										=       restype_atomaa_mask[protein_aatype]

							__snake_case		:      Optional[Any]										=       residx_atomaa_mask
							__snake_case		:      Union[str, Any]										=       residx_atomaa_to_atomaa.long()

							# create the gather indices for mapping back
							__snake_case		:      Dict										=       restype_atomaa_to_atomaa[protein_aatype]
							__snake_case		:      Dict										=       residx_atomaa_to_atomaa.long()

							# create the corresponding mask
							__snake_case		:      List[str]										=       torch.zeros([2_1, 3_7]							,     dtype=torch.floataa							,     device=protein["aatype"].device     )
							for restype, restype_letter in enumerate(rc.restypes     ):
														__snake_case		:      List[str]										=       rc.restype_atoa[restype_letter]
														__snake_case		:      List[Any]										=       rc.residue_atoms[restype_name]
														for atom_name in atom_names:
																					__snake_case		:      Union[str, Any]										=       rc.atom_order[atom_name]
																					__snake_case		:      str										=       1

							__snake_case		:      List[str]										=       restype_atomaa_mask[protein_aatype]
							__snake_case		:      List[str]										=       residx_atomaa_mask

							return protein







def     lowerCAmelCase_					( __lowerCamelCase     ):
							__snake_case		:      str										=       tree_map(lambda __lowerCamelCase     : torch.tensor(__lowerCamelCase							,     device=batch["aatype"].device     )							,     __lowerCamelCase							,     np.ndarray     )
							__snake_case		:      str										=       tensor_tree_map(lambda __lowerCamelCase     : np.array(__lowerCamelCase     )							,     make_atomaa_masks(__lowerCamelCase     )     )
							return out

