



"""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__       (     lowercase_, lowercase_, lowercase_, unittest.TestCase  ):




      """simple docstring"""




      _SCREAMING_SNAKE_CASE      =					StableUnCLIPPipeline
      _SCREAMING_SNAKE_CASE      =					TEXT_TO_IMAGE_PARAMS
      _SCREAMING_SNAKE_CASE      =					TEXT_TO_IMAGE_BATCH_PARAMS
      _SCREAMING_SNAKE_CASE      =					TEXT_TO_IMAGE_IMAGE_PARAMS
      _SCREAMING_SNAKE_CASE      =					TEXT_TO_IMAGE_IMAGE_PARAMS

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



      def 					SCREAMING_SNAKE_CASE__			(				self    :				List[str]	):
        lowerCAmelCase_       :    Tuple								=			3_2
        lowerCAmelCase_       :    List[Any]								=			embedder_hidden_size

        # prior components

        torch.manual_seed(0	)
        lowerCAmelCase_       :    Dict								=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'	)

        torch.manual_seed(0	)
        lowerCAmelCase_       :    Optional[int]								=			CLIPTextModelWithProjection(
            CLIPTextConfig(
                bos_token_id=0   ,      eos_token_id=2   ,      hidden_size=SCREAMING_SNAKE_CASE_   ,      projection_dim=SCREAMING_SNAKE_CASE_   ,      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	)
        lowerCAmelCase_       :    Any								=			PriorTransformer(
            num_attention_heads=2   ,      attention_head_dim=1_2   ,      embedding_dim=SCREAMING_SNAKE_CASE_   ,      num_layers=1   ,      )

        torch.manual_seed(0	)
        lowerCAmelCase_       :    int								=			DDPMScheduler(
            variance_type='fixed_small_log'   ,      prediction_type='sample'   ,      num_train_timesteps=1_0_0_0   ,      clip_sample=SCREAMING_SNAKE_CASE_   ,      clip_sample_range=5.0   ,      beta_schedule='squaredcos_cap_v2'   ,      )

        # regular denoising components

        torch.manual_seed(0	)
        lowerCAmelCase_       :    Any								=			StableUnCLIPImageNormalizer(embedding_dim=SCREAMING_SNAKE_CASE_	)
        lowerCAmelCase_       :    Union[str, Any]								=			DDPMScheduler(beta_schedule='squaredcos_cap_v2'	)

        torch.manual_seed(0	)
        lowerCAmelCase_       :    Union[str, Any]								=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'	)

        torch.manual_seed(0	)
        lowerCAmelCase_       :    Dict								=			CLIPTextModel(
            CLIPTextConfig(
                bos_token_id=0   ,      eos_token_id=2   ,      hidden_size=SCREAMING_SNAKE_CASE_   ,      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	)
        lowerCAmelCase_       :    Dict								=			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=SCREAMING_SNAKE_CASE_   ,      layers_per_block=1   ,      upcast_attention=SCREAMING_SNAKE_CASE_   ,      use_linear_projection=SCREAMING_SNAKE_CASE_   ,      )

        torch.manual_seed(0	)
        lowerCAmelCase_       :    str								=			DDIMScheduler(
            beta_schedule='scaled_linear'   ,      beta_start=0.0_00_85   ,      beta_end=0.0_12   ,      prediction_type='v_prediction'   ,      set_alpha_to_one=SCREAMING_SNAKE_CASE_   ,      steps_offset=1   ,      )

        torch.manual_seed(0	)
        lowerCAmelCase_       :    str								=			AutoencoderKL()

        lowerCAmelCase_       :    Union[str, Any]								=			{
            # 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 					SCREAMING_SNAKE_CASE__			(				self    :				List[Any]   ,      SCREAMING_SNAKE_CASE_    :				List[Any]   ,      SCREAMING_SNAKE_CASE_    :				List[Any]=0	):
        if str(SCREAMING_SNAKE_CASE_	).startswith('mps'	):
          lowerCAmelCase_       :    Optional[int]								=			torch.manual_seed(SCREAMING_SNAKE_CASE_	)
        else:
          lowerCAmelCase_       :    str								=			torch.Generator(device=SCREAMING_SNAKE_CASE_	).manual_seed(SCREAMING_SNAKE_CASE_	)
        lowerCAmelCase_       :    List[str]								=			{
            '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 					SCREAMING_SNAKE_CASE__			(				self    :				Tuple	):
        lowerCAmelCase_       :    List[Any]								=			torch_device == 'cpu'

        self._test_attention_slicing_forward_pass(test_max_difference=SCREAMING_SNAKE_CASE_	)





      def 					SCREAMING_SNAKE_CASE__			(				self    :				Optional[int]	):
        lowerCAmelCase_       :    Union[str, Any]								=			torch_device in ['cpu', 'mps']

        self._test_inference_batch_single_identical(test_max_difference=SCREAMING_SNAKE_CASE_	)







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




      """simple docstring"""




      def 					SCREAMING_SNAKE_CASE__			(				self    :				List[str]	):
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()



      def 					SCREAMING_SNAKE_CASE__			(				self    :				str	):
        lowerCAmelCase_       :    Optional[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'	)

        lowerCAmelCase_       :    Union[str, Any]								=			StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l'   ,      torch_dtype=torch.floataa	)
        pipe.to(SCREAMING_SNAKE_CASE_	)
        pipe.set_progress_bar_config(disable=SCREAMING_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_       :    Optional[int]								=			torch.Generator(device='cpu'	).manual_seed(0	)
        lowerCAmelCase_       :    Optional[Any]								=			pipe('anime turle'   ,      generator=SCREAMING_SNAKE_CASE_   ,      output_type='np'	)

        lowerCAmelCase_       :    Any								=			output.images[0]

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

        assert_mean_pixel_difference(SCREAMING_SNAKE_CASE_   ,      SCREAMING_SNAKE_CASE_	)





      def 					SCREAMING_SNAKE_CASE__			(				self    :				int	):
        torch.cuda.empty_cache()
        torch.cuda.reset_max_memory_allocated()
        torch.cuda.reset_peak_memory_stats()

        lowerCAmelCase_       :    List[str]								=			StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l'   ,      torch_dtype=torch.floataa	)
        lowerCAmelCase_       :    List[str]								=			pipe.to(SCREAMING_SNAKE_CASE_	)
        pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)
        pipe.enable_attention_slicing()
        pipe.enable_sequential_cpu_offload()

        lowerCAmelCase_       :    str								=			pipe(
            'anime turtle'   ,      prior_num_inference_steps=2   ,      num_inference_steps=2   ,      output_type='np'   ,      )

        lowerCAmelCase_       :    Optional[Any]								=			torch.cuda.max_memory_allocated()
        # make sure that less than 7 GB is allocated
        assert mem_bytes < 7 * 1_0**9










"""simple docstring"""







from collections.abc import Iterable
from typing import Generic, TypeVar

lowercase__		:    Any				   =     TypeVar("""_T""")







class 				UpperCamelCase__       (     Generic[_T]  ):




      """simple docstring"""




      def __init__(				self    :				Any   ,      SCREAMING_SNAKE_CASE_    :				Iterable[_T] | None = None	):
        lowerCAmelCase_       :    list[_T]								=			list(iterable or []	)
        lowerCAmelCase_       :    list[_T]								=			[]



      def __len__(				self    :				List[str]	):
        return len(self._stacka	) + len(self._stacka	)



      def __repr__(				self    :				int	):
        return F"Queue({tuple(self._stacka[::-1] + self._stacka	)})"



      def 					SCREAMING_SNAKE_CASE__			(				self    :				Tuple   ,      SCREAMING_SNAKE_CASE_    :				_T	):
        self._stacka.append(SCREAMING_SNAKE_CASE_	)





      def 					SCREAMING_SNAKE_CASE__			(				self    :				str	):
        lowerCAmelCase_       :    List[str]								=			self._stacka.pop
        lowerCAmelCase_       :    int								=			self._stacka.append

        if not self._stacka:
          while self._stacka:
            stacka_append(stacka_pop()	)

        if not self._stacka:
          raise IndexError('Queue is empty'	)
        return self._stacka.pop()


if __name__ == "__main__":
     from doctest import testmod

     testmod()





