




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 						__lowercase		(SCREAMING_SNAKE_CASE__     ,		SCREAMING_SNAKE_CASE__     ,		unittest.TestCase							):




      """simple docstring"""





      _snake_case							     =						AutoencoderKL
      _snake_case							     =						"""sample"""
      _snake_case							     =						1E-2

      @property
      def 	UpperCAmelCase      (      self	)		->				int:
         snake_case      :				Union[str, Any]				=       4
         snake_case      :				List[str]				=       3
         snake_case      :				Tuple				=       (3_2, 3_2)

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

         return {"sample": image}

      @property
      def 	UpperCAmelCase      (      self	)		->				List[str]:
         return (3, 3_2, 3_2)

      @property
      def 	UpperCAmelCase      (      self	)		->				List[str]:
         return (3, 3_2, 3_2)

      def 	UpperCAmelCase      (      self	)		->				int:
         snake_case      :				Optional[Any]				=       {
             """block_out_channels""": [3_2, 6_4],
             """in_channels""": 3,
             """out_channels""": 3,
             """down_block_types""": ["""DownEncoderBlock2D""", """DownEncoderBlock2D"""],
             """up_block_types""": ["""UpDecoderBlock2D""", """UpDecoderBlock2D"""],
             """latent_channels""": 4,
         }
         snake_case      :				str				=       self.dummy_input
         return init_dict, inputs_dict

      def 	UpperCAmelCase      (      self	)		->				Dict:
         pass

      def 	UpperCAmelCase      (      self	)		->				str:
         pass

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

         assert not model.is_gradient_checkpointing and model.training

         snake_case      :				Optional[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      :				int				=       torch.randn_like(A	)
         snake_case      :				str				=       (out - labels).mean()
         loss.backward()

         # re-instantiate the model now enabling gradient checkpointing
         snake_case      :				Dict				=       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      :				int				=       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      :				List[str]				=       (out_a - labels).mean()
         loss_a.backward()

         # compare the output and parameters gradients
         self.assertTrue((loss - loss_a).abs() < 1e-5	)
         snake_case      :				int				=       dict(model.named_parameters()	)
         snake_case      :				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 	UpperCAmelCase      (      self	)		->				int:
         snake_case	,					snake_case      :				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	)
         snake_case      :				List[str]				=       model(**self.dummy_input	)

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

      def 	UpperCAmelCase      (      self	)		->				List[Any]:
         snake_case      :				int				=       AutoencoderKL.from_pretrained("""fusing/autoencoder-kl-dummy"""	)
         snake_case      :				Union[str, Any]				=       model.to(A	)
         model.eval()

         if torch_device == "mps":
            snake_case      :				List[Any]				=       torch.manual_seed(0	)
         else:
            snake_case      :				List[str]				=       torch.Generator(device=A	).manual_seed(0	)

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

         snake_case      :				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":
            snake_case      :				Optional[Any]				=       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":
            snake_case      :				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:
            snake_case      :				Any				=       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 						__lowercase		(unittest.TestCase							):




      """simple docstring"""





      def 	UpperCAmelCase      (      self				,      A				,      A	)		->				Union[str, Any]:
         return f"""gaussian_noise_s={seed}_shape={"_".join([str(A	) for s in shape]	)}.npy"""

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

      def 	UpperCAmelCase      (      self				,      A=0				,      A=(4, 3, 5_1_2, 5_1_2)				,      A=False	)		->				Tuple:
         snake_case      :				Optional[int]				=       torch.floataa if fpaa else torch.floataa
         snake_case      :				Union[str, Any]				=       torch.from_numpy(load_hf_numpy(self.get_file_format(A				,      A	)	)	).to(A	).to(A	)
         return image

      def 	UpperCAmelCase      (      self				,      A="CompVis/stable-diffusion-v1-4"				,      A=False	)		->				int:
         snake_case      :				List[str]				=       """fp16""" if fpaa else None
         snake_case      :				Union[str, Any]				=       torch.floataa if fpaa else torch.floataa

         snake_case      :				List[Any]				=       AutoencoderKL.from_pretrained(
             A				,      subfolder="""vae"""				,      torch_dtype=A				,      revision=A				,      )
         model.to(A	).eval()

         return model

      def 	UpperCAmelCase      (      self				,      A=0	)		->				Optional[Any]:
         if torch_device == "mps":
            return torch.manual_seed(A	)
         return torch.Generator(device=A	).manual_seed(A	)

      @parameterized.expand(
          [
              # fmt: off
              [3_3, [-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]],
              [4_7, [-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 	UpperCAmelCase      (      self				,      A				,      A				,      A	)		->				Optional[Any]:
         snake_case      :				Any				=       self.get_sd_vae_model()
         snake_case      :				Tuple				=       self.get_sd_image(A	)
         snake_case      :				List[Any]				=       self.get_generator(A	)

         with torch.no_grad():
            snake_case      :				str				=       model(A				,      generator=A				,      sample_posterior=A	).sample

         assert sample.shape == image.shape

         snake_case      :				Optional[Any]				=       sample[-1, -2:, -2:, :2].flatten().float().cpu()
         snake_case      :				int				=       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
              [3_3, [-0.05_13, 0.02_89, 1.37_99, 0.21_66, -0.25_73, -0.08_71, 0.51_03, -0.09_99]],
              [4_7, [-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 	UpperCAmelCase      (      self				,      A				,      A	)		->				Optional[Any]:
         snake_case      :				List[str]				=       self.get_sd_vae_model(fpaa=A	)
         snake_case      :				Tuple				=       self.get_sd_image(A				,      fpaa=A	)
         snake_case      :				Tuple				=       self.get_generator(A	)

         with torch.no_grad():
            snake_case      :				Dict				=       model(A				,      generator=A				,      sample_posterior=A	).sample

         assert sample.shape == image.shape

         snake_case      :				str				=       sample[-1, -2:, :2, -2:].flatten().float().cpu()
         snake_case      :				Optional[int]				=       torch.tensor(A	)

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

      @parameterized.expand(
          [
              # fmt: off
              [3_3, [-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]],
              [4_7, [-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 	UpperCAmelCase      (      self				,      A				,      A				,      A	)		->				str:
         snake_case      :				Any				=       self.get_sd_vae_model()
         snake_case      :				List[Any]				=       self.get_sd_image(A	)

         with torch.no_grad():
            snake_case      :				List[str]				=       model(A	).sample

         assert sample.shape == image.shape

         snake_case      :				List[Any]				=       sample[-1, -2:, -2:, :2].flatten().float().cpu()
         snake_case      :				List[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
              [1_3, [-0.20_51, -0.18_03, -0.23_11, -0.21_14, -0.32_92, -0.35_74, -0.29_53, -0.33_23]],
              [3_7, [-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 	UpperCAmelCase      (      self				,      A				,      A	)		->				Dict:
         snake_case      :				Any				=       self.get_sd_vae_model()
         snake_case      :				Tuple				=       self.get_sd_image(A				,      shape=(3, 4, 6_4, 6_4)	)

         with torch.no_grad():
            snake_case      :				Any				=       model.decode(A	).sample

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

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

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

      @parameterized.expand(
          [
              # fmt: off
              [2_7, [-0.03_69, 0.02_07, -0.07_76, -0.06_82, -0.17_47, -0.19_30, -0.14_65, -0.20_39]],
              [1_6, [-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 	UpperCAmelCase      (      self				,      A				,      A	)		->				Optional[int]:
         snake_case      :				List[str]				=       self.get_sd_vae_model(fpaa=A	)
         snake_case      :				List[Any]				=       self.get_sd_image(A				,      shape=(3, 4, 6_4, 6_4)				,      fpaa=A	)

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

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

         snake_case      :				str				=       sample[-1, -2:, :2, -2:].flatten().float().cpu()
         snake_case      :				Optional[int]				=       torch.tensor(A	)

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

      @parameterized.expand([(1_3,), (1_6,), (2_7,)]	)
      @require_torch_gpu
      @unittest.skipIf(not is_xformers_available()				,      reason="""xformers is not required when using PyTorch 2.0."""	)
      def 	UpperCAmelCase      (      self				,      A	)		->				str:
         snake_case      :				Tuple				=       self.get_sd_vae_model(fpaa=A	)
         snake_case      :				Tuple				=       self.get_sd_image(A				,      shape=(3, 4, 6_4, 6_4)				,      fpaa=A	)

         with torch.no_grad():
            snake_case      :				Any				=       model.decode(A	).sample

         model.enable_xformers_memory_efficient_attention()
         with torch.no_grad():
            snake_case      :				Dict				=       model.decode(A	).sample

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

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

      @parameterized.expand([(1_3,), (1_6,), (3_7,)]	)
      @require_torch_gpu
      @unittest.skipIf(not is_xformers_available()				,      reason="""xformers is not required when using PyTorch 2.0."""	)
      def 	UpperCAmelCase      (      self				,      A	)		->				List[Any]:
         snake_case      :				List[str]				=       self.get_sd_vae_model()
         snake_case      :				int				=       self.get_sd_image(A				,      shape=(3, 4, 6_4, 6_4)	)

         with torch.no_grad():
            snake_case      :				Union[str, 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, 5_1_2, 5_1_2]

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

      @parameterized.expand(
          [
              # fmt: off
              [3_3, [-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]],
              [4_7, [-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 	UpperCAmelCase      (      self				,      A				,      A	)		->				int:
         snake_case      :				str				=       self.get_sd_vae_model()
         snake_case      :				List[Any]				=       self.get_sd_image(A	)
         snake_case      :				Optional[Any]				=       self.get_generator(A	)

         with torch.no_grad():
            snake_case      :				List[str]				=       model.encode(A	).latent_dist
            snake_case      :				List[Any]				=       dist.sample(generator=A	)

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

         snake_case      :				Union[str, Any]				=       sample[0, -1, -3:, -3:].flatten().cpu()
         snake_case      :				Tuple				=       torch.tensor(A	)

         snake_case      :				Tuple				=       3e-3 if torch_device != """mps""" else 1e-2
         assert torch_all_close(A				,      A				,      atol=A	)






import os
import unittest

from transformers.models.cpmant.tokenization_cpmant import VOCAB_FILES_NAMES, CpmAntTokenizer
from transformers.testing_utils import require_jieba, tooslow

from ...test_tokenization_common import TokenizerTesterMixin


@require_jieba
class      lowerCamelCase    (SCREAMING_SNAKE_CASE__     ,					unittest.TestCase    ):
  """simple docstring"""







  lowerCamelCase__   =  CpmAntTokenizer
  lowerCamelCase__   =  False







  def       __A   (							self       :				List[str]      )     ->		str:
        super().setUp()

        SCREAMING_SNAKE_CASE_      =     [
            "<d>",
            "</d>",
            "<s>",
            "</s>",
            "</_>",
            "<unk>",
            "<pad>",
            "</n>",
            "我",
            "是",
            "C",
            "P",
            "M",
            "A",
            "n",
            "t",
        ]
        SCREAMING_SNAKE_CASE_      =     os.path.join(self.tmpdirname     ,   VOCAB_FILES_NAMES["vocab_file"]      )
        with open(self.vocab_file     ,   "w"     ,   encoding="utf-8"      ) as vocab_writer:
              vocab_writer.write("".join([x + "\n" for x in vocab_tokens]      )      )







  @tooslow
  def       __A   (							self       :				int      )     ->		Union[str, Any]:
        SCREAMING_SNAKE_CASE_      =     CpmAntTokenizer.from_pretrained("openbmb/cpm-ant-10b"      )
        SCREAMING_SNAKE_CASE_      =     "今天天气真好！"
        SCREAMING_SNAKE_CASE_      =     ["今天", "天气", "真", "好", "！"]
        SCREAMING_SNAKE_CASE_      =     tokenizer.tokenize(__magic_name__      )
        self.assertListEqual(__magic_name__     ,   __magic_name__      )
        SCREAMING_SNAKE_CASE_      =     "今天天气真好！"
        SCREAMING_SNAKE_CASE_      =     [tokenizer.bos_token] + tokens

        SCREAMING_SNAKE_CASE_      =     [6, 9_802, 14_962, 2_082, 831, 244]
        self.assertListEqual(tokenizer.convert_tokens_to_ids(__magic_name__      )     ,   __magic_name__      )

        SCREAMING_SNAKE_CASE_      =     tokenizer.decode(__magic_name__      )
        self.assertEqual(__magic_name__     ,   __magic_name__      )


