
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 						__lowerCAmelCase	(	lowerCAmelCase_ , lowerCAmelCase_ , unittest.TestCase      ):



      """simple docstring"""







      A__	:			List[Any]        =    AutoencoderKL
      A__	:			List[str]        =    '''sample'''
      A__	:			str        =    1E-2




      @property
      def 		snake_case_			(      self		:       Tuple				):
          __lowercase		:			Optional[int]	    =       4
          __lowercase		:			Optional[Any]	    =       3
          __lowercase		:			Any	    =       (32, 32)

          __lowercase		:			Any	    =       floats_tensor((batch_size, num_channels) + sizes				).to(_snake_case				)

          return {"sample": image}




      @property
      def 		snake_case_			(      self		:       Optional[int]				):
          return (3, 32, 32)




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




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




      def 		snake_case_			(      self		:       int				):
          pass




      def 		snake_case_			(      self		:       Dict				):
          pass




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

          assert not model.is_gradient_checkpointing and model.training

          __lowercase		:			Tuple	    =       model(**_snake_case				).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()

          __lowercase		:			Tuple	    =       torch.randn_like(_snake_case				)
          __lowercase		:			Optional[Any]	    =       (out - labels).mean()
          loss.backward()

          # re-instantiate the model now enabling gradient checkpointing
          __lowercase		:			Dict	    =       self.model_class(**_snake_case				)
          # clone model
          model_a.load_state_dict(model.state_dict()				)
          model_a.to(_snake_case				)
          model_a.enable_gradient_checkpointing()

          assert model_a.is_gradient_checkpointing and model_a.training

          __lowercase		:			Optional[Any]	    =       model_a(**_snake_case				).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()
          __lowercase		:			Tuple	    =       (out_a - labels).mean()
          loss_a.backward()

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

          model.to(_snake_case				)
          __lowercase		:			List[Any]	    =       model(**self.dummy_input				)

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




      def 		snake_case_			(      self		:       str				):
          __lowercase		:			Optional[Any]	    =       AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''				)
          __lowercase		:			int	    =       model.to(_snake_case				)
          model.eval()

          if torch_device == "mps":
              __lowercase		:			Optional[int]	    =       torch.manual_seed(0				)
          else:
              __lowercase		:			int	    =       torch.Generator(device=_snake_case				).manual_seed(0				)

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

          __lowercase		:			List[str]	    =       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":
              __lowercase		:			str	    =       torch.tensor(
                  [
                      -4.0_0_7_8E-0_1,
                      -3.8_3_2_3E-0_4,
                      -1.2_6_8_1E-0_1,
                      -1.1_4_6_2E-0_1,
                      2.0_0_9_5E-0_1,
                      1.0_8_9_3E-0_1,
                      -8.8_2_4_7E-0_2,
                      -3.0_3_6_1E-0_1,
                      -9.8_6_4_4E-0_3,
                  ]				)
          elif torch_device == "cpu":
              __lowercase		:			Tuple	    =       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:
              __lowercase		:			List[str]	    =       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(_snake_case				,       _snake_case				,       rtol=1E-2				)				)






@slow
class 						__lowerCAmelCase	(	unittest.TestCase      ):



      """simple docstring"""







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




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




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




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

          __lowercase		:			List[Any]	    =       AutoencoderKL.from_pretrained(
              _snake_case				,       subfolder='''vae'''				,       torch_dtype=_snake_case				,       revision=_snake_case				,       )
          model.to(_snake_case				).eval()

          return model




      def 		snake_case_			(      self		:       Any				,       _snake_case		:       Optional[int]=0				):
          if torch_device == "mps":
              return torch.manual_seed(_snake_case				)
          return torch.Generator(device=_snake_case				).manual_seed(_snake_case				)




      @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		:       int				,       _snake_case		:       str				,       _snake_case		:       Dict				,       _snake_case		:       List[Any]				):
          __lowercase		:			int	    =       self.get_sd_vae_model()
          __lowercase		:			Tuple	    =       self.get_sd_image(_snake_case				)
          __lowercase		:			Any	    =       self.get_generator(_snake_case				)

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

          assert sample.shape == image.shape

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

          assert torch_all_close(_snake_case				,       _snake_case				,       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		:       Optional[Any]				,       _snake_case		:       Optional[Any]				,       _snake_case		:       Dict				):
          __lowercase		:			Optional[int]	    =       self.get_sd_vae_model(fpaa=_snake_case				)
          __lowercase		:			Dict	    =       self.get_sd_image(_snake_case				,       fpaa=_snake_case				)
          __lowercase		:			Union[str, Any]	    =       self.get_generator(_snake_case				)

          with torch.no_grad():
              __lowercase		:			Union[str, Any]	    =       model(_snake_case				,       generator=_snake_case				,       sample_posterior=_snake_case				).sample

          assert sample.shape == image.shape

          __lowercase		:			Tuple	    =       sample[-1, -2:, :2, -2:].flatten().float().cpu()
          __lowercase		:			int	    =       torch.tensor(_snake_case				)

          assert torch_all_close(_snake_case				,       _snake_case				,       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		:       str				,       _snake_case		:       int				,       _snake_case		:       Optional[Any]				,       _snake_case		:       Optional[Any]				):
          __lowercase		:			Optional[Any]	    =       self.get_sd_vae_model()
          __lowercase		:			List[str]	    =       self.get_sd_image(_snake_case				)

          with torch.no_grad():
              __lowercase		:			List[Any]	    =       model(_snake_case				).sample

          assert sample.shape == image.shape

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

          assert torch_all_close(_snake_case				,       _snake_case				,       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		:       Dict				,       _snake_case		:       List[str]				,       _snake_case		:       Tuple				):
          __lowercase		:			Optional[int]	    =       self.get_sd_vae_model()
          __lowercase		:			Optional[Any]	    =       self.get_sd_image(_snake_case				,       shape=(3, 4, 64, 64)				)

          with torch.no_grad():
              __lowercase		:			Optional[int]	    =       model.decode(_snake_case				).sample

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

          __lowercase		:			List[str]	    =       sample[-1, -2:, :2, -2:].flatten().cpu()
          __lowercase		:			Dict	    =       torch.tensor(_snake_case				)

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

          with torch.no_grad():
              __lowercase		:			Any	    =       model.decode(_snake_case				).sample

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

          __lowercase		:			Union[str, Any]	    =       sample[-1, -2:, :2, -2:].flatten().float().cpu()
          __lowercase		:			int	    =       torch.tensor(_snake_case				)

          assert torch_all_close(_snake_case				,       _snake_case				,       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		:       Any				,       _snake_case		:       Any				):
          __lowercase		:			Union[str, Any]	    =       self.get_sd_vae_model(fpaa=_snake_case				)
          __lowercase		:			Optional[Any]	    =       self.get_sd_image(_snake_case				,       shape=(3, 4, 64, 64)				,       fpaa=_snake_case				)

          with torch.no_grad():
              __lowercase		:			str	    =       model.decode(_snake_case				).sample

          model.enable_xformers_memory_efficient_attention()
          with torch.no_grad():
              __lowercase		:			List[str]	    =       model.decode(_snake_case				).sample

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

          assert torch_all_close(_snake_case				,       _snake_case				,       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		:       Any				,       _snake_case		:       Tuple				):
          __lowercase		:			Dict	    =       self.get_sd_vae_model()
          __lowercase		:			int	    =       self.get_sd_image(_snake_case				,       shape=(3, 4, 64, 64)				)

          with torch.no_grad():
              __lowercase		:			int	    =       model.decode(_snake_case				).sample

          model.enable_xformers_memory_efficient_attention()
          with torch.no_grad():
              __lowercase		:			List[Any]	    =       model.decode(_snake_case				).sample

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

          assert torch_all_close(_snake_case				,       _snake_case				,       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		:       Optional[Any]				,       _snake_case		:       Tuple				,       _snake_case		:       Optional[Any]				):
          __lowercase		:			Union[str, Any]	    =       self.get_sd_vae_model()
          __lowercase		:			Tuple	    =       self.get_sd_image(_snake_case				)
          __lowercase		:			Union[str, Any]	    =       self.get_generator(_snake_case				)

          with torch.no_grad():
              __lowercase		:			Dict	    =       model.encode(_snake_case				).latent_dist
              __lowercase		:			Dict	    =       dist.sample(generator=_snake_case				)

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

          __lowercase		:			Optional[Any]	    =       sample[0, -1, -3:, -3:].flatten().cpu()
          __lowercase		:			Optional[int]	    =       torch.tensor(_snake_case				)

          __lowercase		:			str	    =       3E-3 if torch_device != '''mps''' else 1E-2
          assert torch_all_close(_snake_case				,       _snake_case				,       atol=_snake_case				)




import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
 import torch

 from transformers.activations import gelu_new, gelu_python, get_activation


@require_torch
class 						__lowerCAmelCase	(	unittest.TestCase      ):



      """simple docstring"""







      def 		snake_case_			(      self		:       Optional[Any]				):
          __lowercase		:			Union[str, Any]	    =       torch.tensor([-100, -1, -0.1, 0, 0.1, 1.0, 100]				)
          __lowercase		:			Any	    =       get_activation('''gelu'''				)
          self.assertTrue(torch.allclose(gelu_python(_snake_case				)				,       torch_builtin(_snake_case				)				)				)
          self.assertFalse(torch.allclose(gelu_python(_snake_case				)				,       gelu_new(_snake_case				)				)				)




      def 		snake_case_			(      self		:       Dict				):
          __lowercase		:			Any	    =       torch.tensor([-100, -1, -0.1, 0, 0.1, 1.0, 100]				)
          __lowercase		:			List[str]	    =       get_activation('''gelu'''				)
          __lowercase		:			Optional[Any]	    =       get_activation('''gelu_10'''				)

          __lowercase		:			Tuple	    =       torch_builtin(_snake_case				)
          __lowercase		:			List[Any]	    =       geluaa(_snake_case				)

          __lowercase		:			Tuple	    =       torch.where(y_gelu_aa < 10.0				,       1				,       0				)

          self.assertTrue(torch.max(_snake_case				).item() == 10.0				)
          self.assertTrue(torch.allclose(y_gelu * clipped_mask				,       y_gelu_aa * clipped_mask				)				)




      def 		snake_case_			(      self		:       Any				):
          get_activation('''gelu'''				)
          get_activation('''gelu_10'''				)
          get_activation('''gelu_fast'''				)
          get_activation('''gelu_new'''				)
          get_activation('''gelu_python'''				)
          get_activation('''gelu_pytorch_tanh'''				)
          get_activation('''linear'''				)
          get_activation('''mish'''				)
          get_activation('''quick_gelu'''				)
          get_activation('''relu'''				)
          get_activation('''sigmoid'''				)
          get_activation('''silu'''				)
          get_activation('''swish'''				)
          get_activation('''tanh'''				)
          with self.assertRaises(_snake_case				):
              get_activation('''bogus'''				)
          with self.assertRaises(_snake_case				):
              get_activation(_snake_case				)




      def 		snake_case_			(      self		:       Dict				):
          __lowercase		:			Union[str, Any]	    =       get_activation('''gelu'''				)
          __lowercase		:			List[str]	    =       1
          __lowercase		:			Union[str, Any]	    =       get_activation('''gelu'''				)
          self.assertEqual(acta.a				,       1				)
          with self.assertRaises(_snake_case				):
              __lowercase		:			Tuple	    =       acta.a


