



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       _SCREAMING_SNAKE_CASE	(UpperCamelCase			,    UpperCamelCase			,    UpperCamelCase			,    unittest.TestCase ):
     lowerCAmelCase					=			StableUnCLIPPipeline
     lowerCAmelCase					=			TEXT_TO_IMAGE_PARAMS
     lowerCAmelCase					=			TEXT_TO_IMAGE_BATCH_PARAMS
     lowerCAmelCase					=			TEXT_TO_IMAGE_IMAGE_PARAMS
     lowerCAmelCase					=			TEXT_TO_IMAGE_IMAGE_PARAMS

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





     def 	__snake_case    (       self :							int				)->int:
      __SCREAMING_SNAKE_CASE       :    int       =	3_2
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	embedder_hidden_size

      # prior components

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

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    Tuple       =	CLIPTextModelWithProjection(
          CLIPTextConfig(
              bos_token_id=0     ,  eos_token_id=2     ,  hidden_size=UpperCamelCase     ,  projection_dim=UpperCamelCase     ,  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     ,  )				)

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	PriorTransformer(
          num_attention_heads=2     ,  attention_head_dim=1_2     ,  embedding_dim=UpperCamelCase     ,  num_layers=1     ,  )

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    List[Any]       =	DDPMScheduler(
          variance_type="fixed_small_log"     ,  prediction_type="sample"     ,  num_train_timesteps=1_0_0_0     ,  clip_sample=UpperCamelCase     ,  clip_sample_range=5.0     ,  beta_schedule="squaredcos_cap_v2"     ,  )

      # regular denoising components

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    Tuple       =	StableUnCLIPImageNormalizer(embedding_dim=UpperCamelCase				)
      __SCREAMING_SNAKE_CASE       :    Dict       =	DDPMScheduler(beta_schedule="squaredcos_cap_v2"				)

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

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    List[str]       =	CLIPTextModel(
          CLIPTextConfig(
              bos_token_id=0     ,  eos_token_id=2     ,  hidden_size=UpperCamelCase     ,  projection_dim=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     ,  )				)

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    List[str]       =	UNetaDConditionModel(
          sample_size=3_2     ,  in_channels=4     ,  out_channels=4     ,  down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")     ,  up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")     ,  block_out_channels=(3_2, 6_4)     ,  attention_head_dim=(2, 4)     ,  class_embed_type="projection"     ,  projection_class_embeddings_input_dim=embedder_projection_dim * 2     ,  cross_attention_dim=UpperCamelCase     ,  layers_per_block=1     ,  upcast_attention=UpperCamelCase     ,  use_linear_projection=UpperCamelCase     ,  )

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    List[Any]       =	DDIMScheduler(
          beta_schedule="scaled_linear"     ,  beta_start=0.0_0_0_8_5     ,  beta_end=0.0_1_2     ,  prediction_type="v_prediction"     ,  set_alpha_to_one=UpperCamelCase     ,  steps_offset=1     ,  )

      torch.manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	AutoencoderKL()

      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	{
          # 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 	__snake_case    (       self :							Any     ,  UpperCamelCase :							Union[str, Any]     ,  UpperCamelCase :							Any=0				)->List[Any]:
      if str(UpperCamelCase				).startswith("mps"				):
       __SCREAMING_SNAKE_CASE       :    int       =	torch.manual_seed(UpperCamelCase				)
      else:
       __SCREAMING_SNAKE_CASE       :    List[str]       =	torch.Generator(device=UpperCamelCase				).manual_seed(UpperCamelCase				)
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	{
          "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 	__snake_case    (       self :							str				)->Dict:
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	torch_device == "cpu"

      self._test_attention_slicing_forward_pass(test_max_difference=UpperCamelCase				)






     def 	__snake_case    (       self :							Optional[Any]				)->Any:
      __SCREAMING_SNAKE_CASE       :    str       =	torch_device in ["cpu", "mps"]

      self._test_inference_batch_single_identical(test_max_difference=UpperCamelCase				)






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





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





     def 	__snake_case    (       self :							Any				)->List[Any]:
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	load_numpy(
          "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"				)

      __SCREAMING_SNAKE_CASE       :    Dict       =	StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"     ,  torch_dtype=torch.floataa				)
      pipe.to(UpperCamelCase				)
      pipe.set_progress_bar_config(disable=UpperCamelCase				)
      # 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()

      __SCREAMING_SNAKE_CASE       :    Any       =	torch.Generator(device="cpu"				).manual_seed(0				)
      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	pipe("anime turle"     ,  generator=UpperCamelCase     ,  output_type="np"				)

      __SCREAMING_SNAKE_CASE       :    Dict       =	output.images[0]

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

      assert_mean_pixel_difference(UpperCamelCase     ,  UpperCamelCase				)






     def 	__snake_case    (       self :							List[str]				)->Union[str, Any]:
      torch.cuda.empty_cache()
      torch.cuda.reset_max_memory_allocated()
      torch.cuda.reset_peak_memory_stats()

      __SCREAMING_SNAKE_CASE       :    int       =	StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"     ,  torch_dtype=torch.floataa				)
      __SCREAMING_SNAKE_CASE       :    List[Any]       =	pipe.to(UpperCamelCase				)
      pipe.set_progress_bar_config(disable=UpperCamelCase				)
      pipe.enable_attention_slicing()
      pipe.enable_sequential_cpu_offload()

      __SCREAMING_SNAKE_CASE       :    Tuple       =	pipe(
          "anime turtle"     ,  prior_num_inference_steps=2     ,  num_inference_steps=2     ,  output_type="np"     ,  )

      __SCREAMING_SNAKE_CASE       :    int       =	torch.cuda.max_memory_allocated()
      # make sure that less than 7 GB is allocated
      assert mem_bytes < 7 * 1_0**9











from abc import ABC, abstractmethod
from typing import Optional, Union

from .. import Dataset, DatasetDict, Features, IterableDataset, IterableDatasetDict, NamedSplit
from ..utils.typing import NestedDataStructureLike, PathLike

class       _SCREAMING_SNAKE_CASE	(UpperCamelCase ):





     def __init__(       self :							List[str]     ,  UpperCamelCase :							Optional[NestedDataStructureLike[PathLike]] = None     ,  UpperCamelCase :							Optional[NamedSplit] = None     ,  UpperCamelCase :							Optional[Features] = None     ,  UpperCamelCase :							str = None     ,  UpperCamelCase :							bool = False     ,  UpperCamelCase :							bool = False     ,  UpperCamelCase :							Optional[int] = None     ,  **UpperCamelCase :							Dict     ,  )->int:
      __SCREAMING_SNAKE_CASE       :    List[str]       =	path_or_paths
      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	split if split or isinstance(UpperCamelCase     ,  UpperCamelCase				) else "train"
      __SCREAMING_SNAKE_CASE       :    Dict       =	features
      __SCREAMING_SNAKE_CASE       :    List[Any]       =	cache_dir
      __SCREAMING_SNAKE_CASE       :    Tuple       =	keep_in_memory
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	streaming
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	num_proc
      __SCREAMING_SNAKE_CASE       :    int       =	kwargs






     @abstractmethod
     def 	__snake_case    (       self :							Optional[int]				)->Union[Dataset, DatasetDict, IterableDataset, IterableDatasetDict]:
      pass






class       _SCREAMING_SNAKE_CASE	(UpperCamelCase ):





     def __init__(       self :							Optional[Any]     ,  UpperCamelCase :							Optional[Features] = None     ,  UpperCamelCase :							str = None     ,  UpperCamelCase :							bool = False     ,  UpperCamelCase :							bool = False     ,  UpperCamelCase :							Optional[int] = None     ,  **UpperCamelCase :							Dict     ,  )->Optional[int]:
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	features
      __SCREAMING_SNAKE_CASE       :    str       =	cache_dir
      __SCREAMING_SNAKE_CASE       :    Any       =	keep_in_memory
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	streaming
      __SCREAMING_SNAKE_CASE       :    int       =	num_proc
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	kwargs






     @abstractmethod
     def 	__snake_case    (       self :							str				)->Union[Dataset, IterableDataset]:
      pass






