





'''simple docstring'''






import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()




class 			UpperCAmelCase					(      snake_case_							,		snake_case_							,		snake_case_							,		unittest.TestCase       ):
      _lowercase:      Optional[int]											=       StableUnCLIPPipeline
      _lowercase:      Union[str, Any]											=       TEXT_TO_IMAGE_PARAMS
      _lowercase:      List[str]											=       TEXT_TO_IMAGE_BATCH_PARAMS
      _lowercase:      Any											=       TEXT_TO_IMAGE_IMAGE_PARAMS
      _lowercase:      List[Any]											=       TEXT_TO_IMAGE_IMAGE_PARAMS

      # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
      _lowercase:      Tuple											=       False




      def     lowercase__   (      self      :			Tuple  )    ->				Union[str, Any]:
            _lowerCAmelCase												=							32
            _lowerCAmelCase												=							embedder_hidden_size

            # prior components

            torch.manual_seed(0  )
            _lowerCAmelCase												=							CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							CLIPTextModelWithProjection(
                CLIPTextConfig(
                    bos_token_id=0      ,      eos_token_id=2      ,      hidden_size=__snake_case      ,      projection_dim=__snake_case      ,      intermediate_size=37      ,      layer_norm_eps=1E-0_5      ,      num_attention_heads=4      ,      num_hidden_layers=5      ,      pad_token_id=1      ,      vocab_size=10_00      ,      )  )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							PriorTransformer(
                num_attention_heads=2      ,      attention_head_dim=12      ,      embedding_dim=__snake_case      ,      num_layers=1      ,      )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							DDPMScheduler(
                variance_type="""fixed_small_log"""      ,      prediction_type="""sample"""      ,      num_train_timesteps=10_00      ,      clip_sample=__snake_case      ,      clip_sample_range=5.0      ,      beta_schedule="""squaredcos_cap_v2"""      ,      )

            # regular denoising components

            torch.manual_seed(0  )
            _lowerCAmelCase												=							StableUnCLIPImageNormalizer(embedding_dim=__snake_case  )
            _lowerCAmelCase												=							DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""  )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							CLIPTextModel(
                CLIPTextConfig(
                    bos_token_id=0      ,      eos_token_id=2      ,      hidden_size=__snake_case      ,      projection_dim=32      ,      intermediate_size=37      ,      layer_norm_eps=1E-0_5      ,      num_attention_heads=4      ,      num_hidden_layers=5      ,      pad_token_id=1      ,      vocab_size=10_00      ,      )  )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							UNetaDConditionModel(
                sample_size=32      ,      in_channels=4      ,      out_channels=4      ,      down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")      ,      up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")      ,      block_out_channels=(32, 64)      ,      attention_head_dim=(2, 4)      ,      class_embed_type="""projection"""      ,      projection_class_embeddings_input_dim=embedder_projection_dim * 2      ,      cross_attention_dim=__snake_case      ,      layers_per_block=1      ,      upcast_attention=__snake_case      ,      use_linear_projection=__snake_case      ,      )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							DDIMScheduler(
                beta_schedule="""scaled_linear"""      ,      beta_start=0.0_00_85      ,      beta_end=0.0_12      ,      prediction_type="""v_prediction"""      ,      set_alpha_to_one=__snake_case      ,      steps_offset=1      ,      )

            torch.manual_seed(0  )
            _lowerCAmelCase												=							AutoencoderKL()

            _lowerCAmelCase												=							{
                # prior components
                """prior_tokenizer""": prior_tokenizer,
                """prior_text_encoder""": prior_text_encoder,
                """prior""": prior,
                """prior_scheduler""": prior_scheduler,
                # image noising components
                """image_normalizer""": image_normalizer,
                """image_noising_scheduler""": image_noising_scheduler,
                # regular denoising components
                """tokenizer""": tokenizer,
                """text_encoder""": text_encoder,
                """unet""": unet,
                """scheduler""": scheduler,
                """vae""": vae,
            }

            return components




      def     lowercase__   (      self      :			int      ,      __snake_case      :			Optional[int]      ,      __snake_case      :			Dict=0  )    ->				List[str]:
            if str(__snake_case  ).startswith("""mps"""  ):
                  _lowerCAmelCase												=							torch.manual_seed(__snake_case  )
            else:
                  _lowerCAmelCase												=							torch.Generator(device=__snake_case  ).manual_seed(__snake_case  )
            _lowerCAmelCase												=							{
                """prompt""": """A painting of a squirrel eating a burger""",
                """generator""": generator,
                """num_inference_steps""": 2,
                """prior_num_inference_steps""": 2,
                """output_type""": """numpy""",
            }
            return inputs




      def     lowercase__   (      self      :			Union[str, Any]  )    ->				Optional[Any]:
            _lowerCAmelCase												=							torch_device == """cpu"""

            self._test_attention_slicing_forward_pass(test_max_difference=__snake_case  )




      def     lowercase__   (      self      :			Any  )    ->				str:
            _lowerCAmelCase												=							torch_device in ["""cpu""", """mps"""]

            self._test_inference_batch_single_identical(test_max_difference=__snake_case  )




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




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




      def     lowercase__   (      self      :			List[Any]  )    ->				Optional[int]:
            _lowerCAmelCase												=							load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""  )

            _lowerCAmelCase												=							StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""      ,      torch_dtype=torch.floataa  )
            pipe.to(__snake_case  )
            pipe.set_progress_bar_config(disable=__snake_case  )
            # stable unclip will oom when integration tests are run on a V100,
            # so turn on memory savings
            pipe.enable_attention_slicing()
            pipe.enable_sequential_cpu_offload()

            _lowerCAmelCase												=							torch.Generator(device="""cpu"""  ).manual_seed(0  )
            _lowerCAmelCase												=							pipe("""anime turle"""      ,      generator=__snake_case      ,      output_type="""np"""  )

            _lowerCAmelCase												=							output.images[0]

            assert image.shape == (7_68, 7_68, 3)

            assert_mean_pixel_difference(__snake_case      ,      __snake_case  )




      def     lowercase__   (      self      :			Optional[int]  )    ->				int:
            torch.cuda.empty_cache()
            torch.cuda.reset_max_memory_allocated()
            torch.cuda.reset_peak_memory_stats()

            _lowerCAmelCase												=							StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""      ,      torch_dtype=torch.floataa  )
            _lowerCAmelCase												=							pipe.to(__snake_case  )
            pipe.set_progress_bar_config(disable=__snake_case  )
            pipe.enable_attention_slicing()
            pipe.enable_sequential_cpu_offload()

            _lowerCAmelCase												=							pipe(
                """anime turtle"""      ,      prior_num_inference_steps=2      ,      num_inference_steps=2      ,      output_type="""np"""      ,      )

            _lowerCAmelCase												=							torch.cuda.max_memory_allocated()
            # make sure that less than 7 GB is allocated
            assert mem_bytes < 7 * 10**9












'''simple docstring'''






import json
import os
import tempfile

import datasets
from utils import generate_example_dataset, get_duration


A__    :       str              =5_00_00
A__    :       Optional[int]              =50_00

A__       ,						A__    :       Optional[int]              =os.path.split(__file__)
A__    :       Tuple              =os.path.join(RESULTS_BASEPATH, '''results''', RESULTS_FILENAME.replace('''.py''', '''.json'''))
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      for i in range(lowerCAmelCase  ):
            _lowerCAmelCase												=							dataset[i]
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      for i in range(0							,     len(lowerCAmelCase  )							,     lowerCAmelCase  ):
            _lowerCAmelCase												=							dataset[i : i + batch_size]
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      with dataset.formatted_as(type=lowerCAmelCase  ):
            for i in range(lowerCAmelCase  ):
                  _lowerCAmelCase												=							dataset[i]
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      with dataset.formatted_as(type=lowerCAmelCase  ):
            for i in range(0							,     lowerCAmelCase							,     lowerCAmelCase  ):
                  _lowerCAmelCase												=							dataset[i : i + batch_size]
def        UpperCamelCase__				(		):





      """simple docstring"""


      _lowerCAmelCase												=							{"""num examples""": SPEED_TEST_N_EXAMPLES}
      _lowerCAmelCase												=							[
          (read, {"""length""": SMALL_TEST}),
          (read, {"""length""": SPEED_TEST_N_EXAMPLES}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 1_00}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10_00}),
          (read_formatted, {"""type""": """numpy""", """length""": SMALL_TEST}),
          (read_formatted, {"""type""": """pandas""", """length""": SMALL_TEST}),
          (read_formatted, {"""type""": """torch""", """length""": SMALL_TEST}),
          (read_formatted, {"""type""": """tensorflow""", """length""": SMALL_TEST}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10_00}),
      ]

      _lowerCAmelCase												=							[
          (read, {"""length""": SMALL_TEST}),
          (read, {"""length""": SPEED_TEST_N_EXAMPLES}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 1_00}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10_00}),
          (read_formatted, {"""type""": """numpy""", """length""": SMALL_TEST}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10_00}),
      ]
      with tempfile.TemporaryDirectory() as tmp_dir:
            print("""generating dataset"""  )
            _lowerCAmelCase												=							datasets.Features(
                {"""list""": datasets.Sequence(datasets.Value("""float32"""  )  ), """numbers""": datasets.Value("""float32"""  )}  )
            _lowerCAmelCase												=							generate_example_dataset(
                os.path.join(lowerCAmelCase							,     """dataset.arrow"""  )							,     lowerCAmelCase							,     num_examples=lowerCAmelCase							,     seq_shapes={"""list""": (1_00,)}							,     )
            print("""first set of iterations"""  )
            for func, kwargs in functions:
                  print(func.__name__							,     str(lowerCAmelCase  )  )
                  _lowerCAmelCase												=							func(lowerCAmelCase							,     **lowerCAmelCase  )

            print("""shuffling dataset"""  )
            _lowerCAmelCase												=							dataset.shuffle()
            print("""Second set of iterations (after shuffling"""  )
            for func, kwargs in functions_shuffled:
                  print("""shuffled """							,     func.__name__							,     str(lowerCAmelCase  )  )
                  _lowerCAmelCase												=							func(
                      lowerCAmelCase							,     **lowerCAmelCase  )

      with open(lowerCAmelCase							,     """wb"""  ) as f:
            f.write(json.dumps(lowerCAmelCase  ).encode("""utf-8"""  )  )


if __name__ == "__main__":  # useful to run the profiler
      benchmark_iterating()






