




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					(  _UpperCAmelCase			,   _UpperCAmelCase			,   unittest.TestCase      ):
  _UpperCamelCase   :			Union[str, Any] =			AutoencoderKL
  _UpperCamelCase   :			List[Any] =			'sample'
  _UpperCamelCase   :			List[Any] =			1E-2






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






       """simple docstring"""







       lowercase					:   Tuple								= 4
       lowercase					:   str								= 3
       lowercase					:   Optional[Any]								= (32, 32)

       lowercase					:   List[Any]								= floats_tensor((batch_size, num_channels) + sizes      ).to(a__      )

       return {"sample": image}






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






       """simple docstring"""







       return (3, 32, 32)






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






       """simple docstring"""







       return (3, 32, 32)






  def 					__a			(  self		:	int      )		->    List[str]:






       """simple docstring"""







       lowercase					:   List[str]								= {
           '''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					:   Union[str, Any]								= self.dummy_input
       return init_dict, inputs_dict






  def 					__a			(  self		:	Tuple      )		->    int:






       """simple docstring"""







       pass






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






       """simple docstring"""







       pass






  @unittest.skipIf(torch_device == '''mps'''      ,						'''Gradient checkpointing skipped on MPS'''      )
  def 					__a			(  self		:	int      )		->    Dict:






       """simple docstring"""







       lowercase			,       lowercase					:   List[Any]								= self.prepare_init_args_and_inputs_for_common()
       lowercase					:   Union[str, Any]								= self.model_class(**a__      )
       model.to(a__      )

       assert not model.is_gradient_checkpointing and model.training

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

       lowercase					:   Tuple								= torch.randn_like(a__      )
       lowercase					:   List[Any]								= (out - labels).mean()
       loss.backward()

       # re-instantiate the model now enabling gradient checkpointing
       lowercase					:   str								= 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

       lowercase					:   Tuple								= 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()
       lowercase					:   Any								= (out_a - labels).mean()
       loss_a.backward()

       # compare the output and parameters gradients
       self.assertTrue((loss - loss_a).abs() < 1E-5      )
       lowercase					:   Union[str, Any]								= 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 					__a			(  self		:	Tuple      )		->    List[str]:






       """simple docstring"""







       lowercase			,       lowercase					:   str								= 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__      )
       lowercase					:   List[Any]								= model(**self.dummy_input      )

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






  def 					__a			(  self		:	Union[str, Any]      )		->    str:






       """simple docstring"""







       lowercase					:   Any								= AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''      )
       lowercase					:   List[Any]								= model.to(a__      )
       model.eval()

       if torch_device == "mps":
            lowercase					:   Union[str, Any]								= torch.manual_seed(0      )
       else:
            lowercase					:   List[str]								= torch.Generator(device=a__      ).manual_seed(0      )

       lowercase					:   List[str]								= torch.randn(
           1      ,						model.config.in_channels      ,						model.config.sample_size      ,						model.config.sample_size      ,						generator=torch.manual_seed(0      )      ,						)
       lowercase					:   Union[str, Any]								= image.to(a__      )
       with torch.no_grad():
            lowercase					:   Union[str, Any]								= model(a__      ,						sample_posterior=a__      ,						generator=a__      ).sample

       lowercase					:   Dict								= 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_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":
            lowercase					:   int								= torch.tensor(
                [-0.1_352, 0.0_878, 0.0_419, -0.0_818, -0.1_069, 0.0_688, -0.1_458, -0.4_446, -0.0_026]      )
       else:
            lowercase					:   int								= torch.tensor(
                [-0.2_421, 0.4_642, 0.2_507, -0.0_438, 0.0_682, 0.3_160, -0.2_018, -0.0_727, 0.2_485]      )

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


@slow
class 						_A					(  unittest.TestCase      ):






  def 					__a			(  self		:	Tuple      ,						_A		:	Optional[Any]      ,						_A		:	List[str]      )		->    List[str]:






       """simple docstring"""







       return f"""gaussian_noise_s={seed}_shape={'_'.join([str(a__      ) for s in shape]      )}.npy"""






  def 					__a			(  self		:	List[Any]      )		->    Any:






       """simple docstring"""







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






  def 					__a			(  self		:	Optional[int]      ,						_A		:	Optional[Any]=0      ,						_A		:	List[Any]=(4, 3, 512, 512)      ,						_A		:	Optional[Any]=False      )		->    List[str]:






       """simple docstring"""







       lowercase					:   Tuple								= torch.floataa if fpaa else torch.floataa
       lowercase					:   int								= torch.from_numpy(load_hf_numpy(self.get_file_format(a__      ,						a__      )      )      ).to(a__      ).to(a__      )
       return image






  def 					__a			(  self		:	Optional[Any]      ,						_A		:	Dict="CompVis/stable-diffusion-v1-4"      ,						_A		:	List[Any]=False      )		->    Union[str, Any]:






       """simple docstring"""







       lowercase					:   Dict								= '''fp16''' if fpaa else None
       lowercase					:   Tuple								= torch.floataa if fpaa else torch.floataa

       lowercase					:   Union[str, Any]								= AutoencoderKL.from_pretrained(
           a__      ,						subfolder='''vae'''      ,						torch_dtype=a__      ,						revision=a__      ,						)
       model.to(a__      ).eval()

       return model






  def 					__a			(  self		:	Union[str, Any]      ,						_A		:	int=0      )		->    Dict:






       """simple docstring"""







       if torch_device == "mps":
            return torch.manual_seed(a__      )
       return torch.Generator(device=a__      ).manual_seed(a__      )






  @parameterized.expand(
      [
          # fmt: off
          [33, [-0.1_603, 0.9_878, -0.0_495, -0.0_790, -0.2_709, 0.8_375, -0.2_060, -0.0_824], [-0.2_395, 0.0_098, 0.0_102, -0.0_709, -0.2_840, -0.0_274, -0.0_718, -0.1_824]],
          [47, [-0.2_376, 0.1_168, 0.1_332, -0.4_840, -0.2_508, -0.0_791, -0.0_493, -0.4_089], [0.0_350, 0.0_847, 0.0_467, 0.0_344, -0.0_842, -0.0_547, -0.0_633, -0.1_131]],
          # fmt: on
      ]      )
  def 					__a			(  self		:	Optional[int]      ,						_A		:	int      ,						_A		:	Union[str, Any]      ,						_A		:	Optional[int]      )		->    Optional[int]:






       """simple docstring"""







       lowercase					:   List[str]								= self.get_sd_vae_model()
       lowercase					:   List[str]								= self.get_sd_image(a__      )
       lowercase					:   str								= self.get_generator(a__      )

       with torch.no_grad():
            lowercase					:   str								= model(a__      ,						generator=a__      ,						sample_posterior=a__      ).sample

       assert sample.shape == image.shape

       lowercase					:   List[Any]								= sample[-1, -2:, -2:, :2].flatten().float().cpu()
       lowercase					:   List[Any]								= 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.0_513, 0.0_289, 1.3_799, 0.2_166, -0.2_573, -0.0_871, 0.5_103, -0.0_999]],
          [47, [-0.4_128, -0.1_320, -0.3_704, 0.1_965, -0.4_116, -0.2_332, -0.3_340, 0.2_247]],
          # fmt: on
      ]      )
  @require_torch_gpu
  def 					__a			(  self		:	Union[str, Any]      ,						_A		:	Union[str, Any]      ,						_A		:	str      )		->    Dict:






       """simple docstring"""







       lowercase					:   Optional[Any]								= self.get_sd_vae_model(fpaa=a__      )
       lowercase					:   Any								= self.get_sd_image(a__      ,						fpaa=a__      )
       lowercase					:   List[Any]								= self.get_generator(a__      )

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

       assert sample.shape == image.shape

       lowercase					:   Tuple								= sample[-1, -2:, :2, -2:].flatten().float().cpu()
       lowercase					:   Optional[Any]								= torch.tensor(a__      )

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






  @parameterized.expand(
      [
          # fmt: off
          [33, [-0.1_609, 0.9_866, -0.0_487, -0.0_777, -0.2_716, 0.8_368, -0.2_055, -0.0_814], [-0.2_395, 0.0_098, 0.0_102, -0.0_709, -0.2_840, -0.0_274, -0.0_718, -0.1_824]],
          [47, [-0.2_377, 0.1_147, 0.1_333, -0.4_841, -0.2_506, -0.0_805, -0.0_491, -0.4_085], [0.0_350, 0.0_847, 0.0_467, 0.0_344, -0.0_842, -0.0_547, -0.0_633, -0.1_131]],
          # fmt: on
      ]      )
  def 					__a			(  self		:	Optional[Any]      ,						_A		:	str      ,						_A		:	Tuple      ,						_A		:	List[Any]      )		->    Optional[Any]:






       """simple docstring"""







       lowercase					:   Tuple								= self.get_sd_vae_model()
       lowercase					:   List[Any]								= self.get_sd_image(a__      )

       with torch.no_grad():
            lowercase					:   Dict								= model(a__      ).sample

       assert sample.shape == image.shape

       lowercase					:   Dict								= sample[-1, -2:, -2:, :2].flatten().float().cpu()
       lowercase					:   str								= 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.2_051, -0.1_803, -0.2_311, -0.2_114, -0.3_292, -0.3_574, -0.2_953, -0.3_323]],
          [37, [-0.2_632, -0.2_625, -0.2_199, -0.2_741, -0.4_539, -0.4_990, -0.3_720, -0.4_925]],
          # fmt: on
      ]      )
  @require_torch_gpu
  def 					__a			(  self		:	str      ,						_A		:	Optional[int]      ,						_A		:	Any      )		->    int:






       """simple docstring"""







       lowercase					:   Dict								= self.get_sd_vae_model()
       lowercase					:   Union[str, Any]								= self.get_sd_image(a__      ,						shape=(3, 4, 64, 64)      )

       with torch.no_grad():
            lowercase					:   List[str]								= model.decode(a__      ).sample

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

       lowercase					:   List[str]								= sample[-1, -2:, :2, -2:].flatten().cpu()
       lowercase					:   int								= torch.tensor(a__      )

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






  @parameterized.expand(
      [
          # fmt: off
          [27, [-0.0_369, 0.0_207, -0.0_776, -0.0_682, -0.1_747, -0.1_930, -0.1_465, -0.2_039]],
          [16, [-0.1_628, -0.2_134, -0.2_747, -0.2_642, -0.3_774, -0.4_404, -0.3_687, -0.4_277]],
          # fmt: on
      ]      )
  @require_torch_gpu
  def 					__a			(  self		:	List[Any]      ,						_A		:	Any      ,						_A		:	Optional[int]      )		->    List[Any]:






       """simple docstring"""







       lowercase					:   Optional[int]								= self.get_sd_vae_model(fpaa=a__      )
       lowercase					:   Tuple								= self.get_sd_image(a__      ,						shape=(3, 4, 64, 64)      ,						fpaa=a__      )

       with torch.no_grad():
            lowercase					:   Union[str, Any]								= model.decode(a__      ).sample

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

       lowercase					:   Any								= sample[-1, -2:, :2, -2:].flatten().float().cpu()
       lowercase					:   Dict								= 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 					__a			(  self		:	Tuple      ,						_A		:	Any      )		->    int:






       """simple docstring"""







       lowercase					:   Tuple								= self.get_sd_vae_model(fpaa=a__      )
       lowercase					:   Optional[int]								= self.get_sd_image(a__      ,						shape=(3, 4, 64, 64)      ,						fpaa=a__      )

       with torch.no_grad():
            lowercase					:   List[Any]								= model.decode(a__      ).sample

       model.enable_xformers_memory_efficient_attention()
       with torch.no_grad():
            lowercase					:   str								= 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 					__a			(  self		:	int      ,						_A		:	Optional[int]      )		->    str:






       """simple docstring"""







       lowercase					:   List[str]								= self.get_sd_vae_model()
       lowercase					:   str								= self.get_sd_image(a__      ,						shape=(3, 4, 64, 64)      )

       with torch.no_grad():
            lowercase					:   List[Any]								= model.decode(a__      ).sample

       model.enable_xformers_memory_efficient_attention()
       with torch.no_grad():
            lowercase					:   List[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.3_001, 0.0_918, -2.6_984, -3.9_720, -3.2_099, -5.0_353, 1.7_338, -0.2_065, 3.4_267]],
          [47, [-1.5_030, -4.3_871, -6.0_355, -9.1_157, -1.6_661, -2.7_853, 2.1_607, -5.0_823, 2.5_633]],
          # fmt: on
      ]      )
  def 					__a			(  self		:	str      ,						_A		:	Optional[Any]      ,						_A		:	List[str]      )		->    List[str]:






       """simple docstring"""







       lowercase					:   Optional[Any]								= self.get_sd_vae_model()
       lowercase					:   str								= self.get_sd_image(a__      )
       lowercase					:   Dict								= self.get_generator(a__      )

       with torch.no_grad():
            lowercase					:   int								= model.encode(a__      ).latent_dist
            lowercase					:   Dict								= dist.sample(generator=a__      )

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

       lowercase					:   List[Any]								= sample[0, -1, -3:, -3:].flatten().cpu()
       lowercase					:   Optional[Any]								= torch.tensor(a__      )

       lowercase					:   List[str]								= 3E-3 if torch_device != '''mps''' else 1E-2
       assert torch_all_close(a__      ,						a__      ,						atol=a__      )






import os
import unittest

from transformers.models.bartpho.tokenization_bartpho import VOCAB_FILES_NAMES, BartphoTokenizer
from transformers.testing_utils import get_tests_dir

from ...test_tokenization_common import TokenizerTesterMixin


snake_case_				=			get_tests_dir('fixtures/test_sentencepiece_bpe.model')

class 		SCREAMING_SNAKE_CASE__				(       _UpperCAmelCase						,      unittest.TestCase  ):
      A_						:				List[Any]							=    BartphoTokenizer
      A_						:				List[str]							=    False
      A_						:				Optional[Any]							=    True


      def    a (self       :      Tuple      ):





         """simple docstring"""



         super().setUp()

         __snake_case							       =   ['''▁This''', '''▁is''', '''▁a''', '''▁t''', '''est''']
         __snake_case							       =   dict(zip(a__				,			range(len(a__      )      )      )      )
         __snake_case							       =   {'''unk_token''': '''<unk>'''}

         __snake_case							       =   os.path.join(self.tmpdirname				,			VOCAB_FILES_NAMES['''monolingual_vocab_file''']      )
         with open(self.monolingual_vocab_file				,			'''w'''				,			encoding='''utf-8'''      ) as fp:
            for token in vocab_tokens:
               fp.write(f"""{token} {vocab_tokens[token]}\n"""      )

         __snake_case							       =   BartphoTokenizer(a__				,			self.monolingual_vocab_file				,			**self.special_tokens_map      )
         tokenizer.save_pretrained(self.tmpdirname      )


      def    a (self       :      str				,			**a__       :      str      ):





         """simple docstring"""



         kwargs.update(self.special_tokens_map      )
         return BartphoTokenizer.from_pretrained(self.tmpdirname				,			**a__      )


      def    a (self       :      str				,			a__       :      Any      ):





         """simple docstring"""



         __snake_case							       =   '''This is a là test'''
         __snake_case							       =   '''This is a<unk><unk> test'''
         return input_text, output_text


      def    a (self       :      Dict      ):





         """simple docstring"""



         __snake_case							       =   BartphoTokenizer(a__				,			self.monolingual_vocab_file				,			**self.special_tokens_map      )
         __snake_case							       =   '''This is a là test'''
         __snake_case							       =   '''▁This ▁is ▁a ▁l à ▁t est'''.split()
         __snake_case							       =   tokenizer.tokenize(a__      )
         self.assertListEqual(a__				,			a__      )

         __snake_case							       =   tokens + [tokenizer.unk_token]
         __snake_case							       =   [4, 5, 6, 3, 3, 7, 8, 3]
         self.assertListEqual(tokenizer.convert_tokens_to_ids(a__      )				,			a__      )

