
'''simple docstring'''

import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class 	__lowerCAmelCase    (  __a       ,					__a       ,					unittest.TestCase				):
     snake_case   :    Optional[Any]							      =      StableDiffusionParadigmsPipeline
     snake_case   :    Any							      =      TEXT_TO_IMAGE_PARAMS
     snake_case   :    Any							      =      TEXT_TO_IMAGE_BATCH_PARAMS
     snake_case   :    Dict							      =      TEXT_TO_IMAGE_IMAGE_PARAMS
     snake_case   :    Any							      =      TEXT_TO_IMAGE_IMAGE_PARAMS



     def     snake_case_ (self	):
          torch.manual_seed(0	)
          _UpperCAmelCase					:  Tuple         =    UNetaDConditionModel(
              block_out_channels=(3_2, 6_4)						,	layers_per_block=2						,	sample_size=3_2						,	in_channels=4						,	out_channels=4						,	down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")						,	up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")						,	cross_attention_dim=3_2						,	attention_head_dim=(2, 4)						,	use_linear_projection=lowerCAmelCase__						,	)
          _UpperCAmelCase					:  Union[str, Any]         =    DDIMParallelScheduler(
              beta_start=0.0_0_0_8_5						,	beta_end=0.0_1_2						,	beta_schedule="""scaled_linear"""						,	clip_sample=lowerCAmelCase__						,	set_alpha_to_one=lowerCAmelCase__						,	)
          torch.manual_seed(0	)
          _UpperCAmelCase					:  str         =    AutoencoderKL(
              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						,	sample_size=1_2_8						,	)
          torch.manual_seed(0	)
          _UpperCAmelCase					:  Any         =    CLIPTextConfig(
              bos_token_id=0						,	eos_token_id=2						,	hidden_size=3_2						,	intermediate_size=3_7						,	layer_norm_eps=1e-05						,	num_attention_heads=4						,	num_hidden_layers=5						,	pad_token_id=1						,	vocab_size=1_0_0_0						,	hidden_act="""gelu"""						,	projection_dim=5_1_2						,	)
          _UpperCAmelCase					:  List[Any]         =    CLIPTextModel(lowerCAmelCase__	)
          _UpperCAmelCase					:  Tuple         =    CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""	)

          _UpperCAmelCase					:  Optional[int]         =    {
              """unet""": unet,
              """scheduler""": scheduler,
              """vae""": vae,
              """text_encoder""": text_encoder,
              """tokenizer""": tokenizer,
              """safety_checker""": None,
              """feature_extractor""": None,
          }
          return components



     def     snake_case_ (self						,	lowerCAmelCase__						,	lowerCAmelCase__=0	):
          if str(lowerCAmelCase__	).startswith("""mps"""	):
               _UpperCAmelCase					:  Optional[int]         =    torch.manual_seed(lowerCAmelCase__	)
          else:
               _UpperCAmelCase					:  int         =    torch.Generator(device=lowerCAmelCase__	).manual_seed(lowerCAmelCase__	)
          _UpperCAmelCase					:  str         =    {
              """prompt""": """a photograph of an astronaut riding a horse""",
              """generator""": generator,
              """num_inference_steps""": 1_0,
              """guidance_scale""": 6.0,
              """output_type""": """numpy""",
              """parallel""": 3,
              """debug""": True,
          }
          return inputs



     def     snake_case_ (self	):
          _UpperCAmelCase					:  List[str]         =    """cpu"""  # ensure determinism for the device-dependent torch.Generator
          _UpperCAmelCase					:  Optional[int]         =    self.get_dummy_components()
          _UpperCAmelCase					:  List[str]         =    StableDiffusionParadigmsPipeline(**lowerCAmelCase__	)
          _UpperCAmelCase					:  Optional[Any]         =    sd_pipe.to(lowerCAmelCase__	)
          sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__	)

          _UpperCAmelCase					:  Optional[Any]         =    self.get_dummy_inputs(lowerCAmelCase__	)
          _UpperCAmelCase					:  Tuple         =    sd_pipe(**lowerCAmelCase__	).images
          _UpperCAmelCase					:  Tuple         =    image[0, -3:, -3:, -1]
          assert image.shape == (1, 6_4, 6_4, 3)

          _UpperCAmelCase					:  Dict         =    np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]	)

          assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-2



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Dict         =    """cpu"""  # ensure determinism for the device-dependent torch.Generator
          _UpperCAmelCase					:  List[str]         =    self.get_dummy_components()
          torch.manual_seed(0	)
          _UpperCAmelCase					:  Tuple         =    DDPMParallelScheduler()
          torch.manual_seed(0	)
          _UpperCAmelCase					:  Optional[Any]         =    StableDiffusionParadigmsPipeline(**lowerCAmelCase__	)
          _UpperCAmelCase					:  Union[str, Any]         =    sd_pipe.to(lowerCAmelCase__	)
          sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__	)

          _UpperCAmelCase					:  Union[str, Any]         =    self.get_dummy_inputs(lowerCAmelCase__	)
          _UpperCAmelCase					:  int         =    sd_pipe(**lowerCAmelCase__	).images
          _UpperCAmelCase					:  int         =    image[0, -3:, -3:, -1]
          assert image.shape == (1, 6_4, 6_4, 3)

          _UpperCAmelCase					:  Tuple         =    np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]	)

          assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-2



     def     snake_case_ (self	):
          super().test_inference_batch_consistent(batch_sizes=[1, 2]	)



     def     snake_case_ (self	):
          super().test_inference_batch_single_identical(batch_size=2						,	expected_max_diff=3e-3	)



     def     snake_case_ (self	):
          _UpperCAmelCase					:  str         =    """cpu"""  # ensure determinism for the device-dependent torch.Generator
          _UpperCAmelCase					:  Optional[Any]         =    self.get_dummy_components()
          _UpperCAmelCase					:  Optional[Any]         =    StableDiffusionParadigmsPipeline(**lowerCAmelCase__	)
          _UpperCAmelCase					:  Optional[int]         =    sd_pipe.to(lowerCAmelCase__	)
          sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__	)

          _UpperCAmelCase					:  Dict         =    self.get_dummy_inputs(lowerCAmelCase__	)
          _UpperCAmelCase					:  int         =    """french fries"""
          _UpperCAmelCase					:  Optional[Any]         =    sd_pipe(**lowerCAmelCase__						,	negative_prompt=lowerCAmelCase__	)
          _UpperCAmelCase					:  int         =    output.images
          _UpperCAmelCase					:  Tuple         =    image[0, -3:, -3:, -1]

          assert image.shape == (1, 6_4, 6_4, 3)

          _UpperCAmelCase					:  int         =    np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]	)

          assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-2







@slow
@require_torch_gpu
class 	__lowerCAmelCase    (  unittest.TestCase				):



     def     snake_case_ (self	):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()



     def     snake_case_ (self						,	lowerCAmelCase__=0	):
          _UpperCAmelCase					:  Any         =    torch.Generator(device=lowerCAmelCase__	).manual_seed(lowerCAmelCase__	)
          _UpperCAmelCase					:  int         =    {
              """prompt""": """a photograph of an astronaut riding a horse""",
              """generator""": generator,
              """num_inference_steps""": 1_0,
              """guidance_scale""": 7.5,
              """output_type""": """numpy""",
              """parallel""": 3,
              """debug""": True,
          }
          return inputs



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Any         =    """stabilityai/stable-diffusion-2-base"""
          _UpperCAmelCase					:  int         =    DDIMParallelScheduler.from_pretrained(lowerCAmelCase__						,	subfolder="""scheduler"""	)
          _UpperCAmelCase					:  str         =    StableDiffusionParadigmsPipeline.from_pretrained(lowerCAmelCase__						,	scheduler=lowerCAmelCase__						,	safety_checker=lowerCAmelCase__	)
          pipe.to(lowerCAmelCase__	)
          pipe.set_progress_bar_config(disable=lowerCAmelCase__	)
          pipe.enable_attention_slicing()

          _UpperCAmelCase					:  List[Any]         =    self.get_inputs()
          _UpperCAmelCase					:  Union[str, Any]         =    pipe(**lowerCAmelCase__	).images
          _UpperCAmelCase					:  int         =    image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 5_1_2, 5_1_2, 3)

          _UpperCAmelCase					:  Union[str, Any]         =    np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]	)

          assert np.abs(expected_slice - image_slice	).max() < 1e-2




'''simple docstring'''

import concurrent.futures
import json
import os
import shutil
import tempfile
import unittest

from transformers import AutoTokenizer, PreTrainedTokenizerFast
from transformers.testing_utils import require_tokenizers

from ..test_tokenization_common import TokenizerTesterMixin







@require_tokenizers
class 	__lowerCAmelCase    (  __a       ,					unittest.TestCase				):
     snake_case   :    Dict							      =      PreTrainedTokenizerFast
     snake_case   :    Optional[Any]							      =      False
     snake_case   :    List[str]							      =      True
     snake_case   :    Any							      =      """tokenizer_file"""



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Dict         =    False  # because we don't have pretrained_vocab_files_map
          super().setUp()
          _UpperCAmelCase					:  Optional[Any]         =    True

          _UpperCAmelCase					:  int         =    ["""robot-test/dummy-tokenizer-fast""", """robot-test/dummy-tokenizer-wordlevel"""]
          _UpperCAmelCase					:  Union[str, Any]         =    """SaulLu/dummy-tokenizer-bytelevel-bpe"""

          # Inclusion of 2 tokenizers to test different types of models (Unigram and WordLevel for the moment)
          _UpperCAmelCase					:  List[str]         =    [(PreTrainedTokenizerFast, model_path, {}) for model_path in model_paths]

          _UpperCAmelCase					:  Optional[Any]         =    PreTrainedTokenizerFast.from_pretrained(model_paths[0]	)
          tokenizer.save_pretrained(self.tmpdirname	)



     def     snake_case_ (self	):
          # We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
          # model
          pass



     def     snake_case_ (self	):
          # We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
          # model
          pass



     def     snake_case_ (self	):
          # We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
          # model
          pass



     def     snake_case_ (self	):
          # PreTrainedTokenizerFast doesn't have tokenizer_file in its signature
          pass



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Optional[Any]         =    self.tmpdirname
          # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
          for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
               with self.subTest(F"{tokenizer.__class__.__name__} ({pretrained_name})"	):
                    try:
                         _UpperCAmelCase					:  Optional[int]         =    tempfile.mkdtemp()
                         _UpperCAmelCase					:  Tuple         =    self.rust_tokenizer_class.from_pretrained(lowerCAmelCase__						,	**lowerCAmelCase__	)

                         tokenizer.save_pretrained(self.tmpdirname	)
                         super().test_training_new_tokenizer()
                    finally:
                         # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                         # is restored
                         shutil.rmtree(self.tmpdirname	)
                         _UpperCAmelCase					:  Optional[int]         =    tmpdirname_orig



     def     snake_case_ (self	):
          _UpperCAmelCase					:  int         =    self.tmpdirname
          # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
          for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
               with self.subTest(F"{tokenizer.__class__.__name__} ({pretrained_name})"	):
                    try:
                         _UpperCAmelCase					:  Any         =    tempfile.mkdtemp()
                         _UpperCAmelCase					:  Union[str, Any]         =    self.rust_tokenizer_class.from_pretrained(lowerCAmelCase__						,	**lowerCAmelCase__	)

                         tokenizer.save_pretrained(self.tmpdirname	)
                         super().test_training_new_tokenizer_with_special_tokens_change()
                    finally:
                         # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                         # is restored
                         shutil.rmtree(self.tmpdirname	)
                         _UpperCAmelCase					:  Dict         =    tmpdirname_orig



     def     snake_case_ (self	):
          _UpperCAmelCase					:  List[Any]         =    self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name	)

          _UpperCAmelCase					:  Dict         =    ("""a""" for _ in range(1_0_0_0	))
          _UpperCAmelCase					:  Optional[int]         =    tokenizer.train_new_from_iterator(text_iterator=lowerCAmelCase__						,	length=1_0_0_0						,	vocab_size=5_0	)

          _UpperCAmelCase					:  int         =    new_tokenizer.encode("""a🤗"""	)
          self.assertEqual(lowerCAmelCase__						,	[6_4, 1_7_2, 2_5_3, 9_7, 2_4_5]	)







@require_tokenizers
class 	__lowerCAmelCase    (  unittest.TestCase				):



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Any         =    AutoTokenizer.from_pretrained("""bert-base-cased"""	)
          _UpperCAmelCase					:  Dict         =    json.loads(tokenizer._tokenizer.to_str()	)
          _UpperCAmelCase					:  List[Any]         =    len(lowerCAmelCase__	)

          with tempfile.TemporaryDirectory() as tmp_dir:
               # Hack to save this in the tokenizer_config.json
               _UpperCAmelCase					:  Union[str, Any]         =    ["""tokenizer.4.0.0.json"""]
               tokenizer.save_pretrained(lowerCAmelCase__	)
               json.dump(lowerCAmelCase__						,	open(os.path.join(lowerCAmelCase__						,	"""tokenizer.4.0.0.json"""	)						,	"""w"""	)	)

               # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
               _UpperCAmelCase					:  Optional[int]         =    AutoTokenizer.from_pretrained(lowerCAmelCase__	)
               self.assertEqual(len(lowerCAmelCase__	)						,	len(lowerCAmelCase__	) + 1	)
               _UpperCAmelCase					:  Tuple         =    json.loads(new_tokenizer._tokenizer.to_str()	)
               self.assertIn("""huggingface"""						,	json_tokenizer["""model"""]["""vocab"""]	)

               # Will need to be adjusted if we reach v42 and this test is still here.
               # Should pick the old tokenizer file as the version of Transformers is < 4.0.0
               shutil.move(os.path.join(lowerCAmelCase__						,	"""tokenizer.4.0.0.json"""	)						,	os.path.join(lowerCAmelCase__						,	"""tokenizer.42.0.0.json"""	)	)
               _UpperCAmelCase					:  Tuple         =    ["""tokenizer.42.0.0.json"""]
               tokenizer.save_pretrained(lowerCAmelCase__	)
               _UpperCAmelCase					:  str         =    AutoTokenizer.from_pretrained(lowerCAmelCase__	)
               self.assertEqual(len(lowerCAmelCase__	)						,	len(lowerCAmelCase__	)	)
               _UpperCAmelCase					:  Union[str, Any]         =    json.loads(new_tokenizer._tokenizer.to_str()	)
               self.assertNotIn("""huggingface"""						,	json_tokenizer["""model"""]["""vocab"""]	)



     def     snake_case_ (self	):
          # This repo has two tokenizer files, one for v4.0.0 and above with an added token, one for versions lower.
          _UpperCAmelCase					:  List[Any]         =    """hf-internal-testing/test-two-tokenizers"""

          # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
          _UpperCAmelCase					:  Dict         =    AutoTokenizer.from_pretrained(lowerCAmelCase__	)
          self.assertEqual(len(lowerCAmelCase__	)						,	2_8_9_9_7	)
          _UpperCAmelCase					:  List[str]         =    json.loads(tokenizer._tokenizer.to_str()	)
          self.assertIn("""huggingface"""						,	json_tokenizer["""model"""]["""vocab"""]	)

          # Testing an older version by monkey-patching the version in the module it's used.
          import transformers as old_transformers

          _UpperCAmelCase					:  Optional[Any]         =    """3.0.0"""
          _UpperCAmelCase					:  str         =    old_transformers.models.auto.AutoTokenizer.from_pretrained(lowerCAmelCase__	)
          self.assertEqual(len(lowerCAmelCase__	)						,	2_8_9_9_6	)
          _UpperCAmelCase					:  Optional[int]         =    json.loads(old_tokenizer._tokenizer.to_str()	)
          self.assertNotIn("""huggingface"""						,	json_tokenizer["""model"""]["""vocab"""]	)







@require_tokenizers
class 	__lowerCAmelCase    (  unittest.TestCase				):



     def     snake_case_ (self	):
          # See https://github.com/huggingface/transformers/pull/12550
          # and https://github.com/huggingface/tokenizers/issues/537
          _UpperCAmelCase					:  int         =    PreTrainedTokenizerFast.from_pretrained("""robot-test/dummy-tokenizer-wordlevel"""	)
          _UpperCAmelCase					:  str         =    """The Matrix is a 1999 science fiction action film."""

          with concurrent.futures.ThreadPoolExecutor() as executor:
               _UpperCAmelCase					:  List[Any]         =    [executor.submit(self.fetch						,	lowerCAmelCase__						,	lowerCAmelCase__	) for i in range(1_0	)]
               _UpperCAmelCase					:  Any         =    [future.result() for future in futures]
               self.assertEqual(lowerCAmelCase__						,	[[1, 1_0, 0, 8, 0, 1_8, 0, 0, 0, 2] for i in range(1_0	)]	)



     def     snake_case_ (self						,	lowerCAmelCase__						,	lowerCAmelCase__	):
          return tokenizer.encode(lowerCAmelCase__						,	truncation="""longest_first"""						,	padding="""longest"""	)


