





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







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__							:Tuple	     =					StableUnCLIPPipeline
  SCREAMING_SNAKE_CASE__							:Union[str, Any]	     =					TEXT_TO_IMAGE_PARAMS
  SCREAMING_SNAKE_CASE__							:List[str]	     =					TEXT_TO_IMAGE_BATCH_PARAMS
  SCREAMING_SNAKE_CASE__							:Any	     =					TEXT_TO_IMAGE_IMAGE_PARAMS
  SCREAMING_SNAKE_CASE__							:str	     =					TEXT_TO_IMAGE_IMAGE_PARAMS

  # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
  SCREAMING_SNAKE_CASE__							:Union[str, Any]	     =					False

  def    __SCREAMING_SNAKE_CASE				(		self   :       str	)		->				Tuple:
        _UpperCamelCase      :	Tuple						=       32
        _UpperCamelCase      :	List[str]						=       embedder_hidden_size

        # prior components

        torch.manual_seed(0	)
        _UpperCamelCase      :	Optional[int]						=       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"	)

        torch.manual_seed(0	)
        _UpperCamelCase      :	int						=       CLIPTextModelWithProjection(
            CLIPTextConfig(
                bos_token_id=0					,   eos_token_id=2					,   hidden_size=__a					,   projection_dim=__a					,   intermediate_size=37					,   layer_norm_eps=1e-0_5					,   num_attention_heads=4					,   num_hidden_layers=5					,   pad_token_id=1					,   vocab_size=1000					,   )	)

        torch.manual_seed(0	)
        _UpperCamelCase      :	str						=       PriorTransformer(
            num_attention_heads=2					,   attention_head_dim=12					,   embedding_dim=__a					,   num_layers=1					,   )

        torch.manual_seed(0	)
        _UpperCamelCase      :	List[Any]						=       DDPMScheduler(
            variance_type="fixed_small_log"					,   prediction_type="sample"					,   num_train_timesteps=1000					,   clip_sample=__a					,   clip_sample_range=5.0					,   beta_schedule="squaredcos_cap_v2"					,   )

        # regular denoising components

        torch.manual_seed(0	)
        _UpperCamelCase      :	str						=       StableUnCLIPImageNormalizer(embedding_dim=__a	)
        _UpperCamelCase      :	Optional[Any]						=       DDPMScheduler(beta_schedule="squaredcos_cap_v2"	)

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

        torch.manual_seed(0	)
        _UpperCamelCase      :	int						=       CLIPTextModel(
            CLIPTextConfig(
                bos_token_id=0					,   eos_token_id=2					,   hidden_size=__a					,   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=1000					,   )	)

        torch.manual_seed(0	)
        _UpperCamelCase      :	Union[str, Any]						=       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=__a					,   layers_per_block=1					,   upcast_attention=__a					,   use_linear_projection=__a					,   )

        torch.manual_seed(0	)
        _UpperCamelCase      :	Optional[Any]						=       DDIMScheduler(
            beta_schedule="scaled_linear"					,   beta_start=0.0_00_85					,   beta_end=0.0_12					,   prediction_type="v_prediction"					,   set_alpha_to_one=__a					,   steps_offset=1					,   )

        torch.manual_seed(0	)
        _UpperCamelCase      :	Union[str, Any]						=       AutoencoderKL()

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

        self._test_attention_slicing_forward_pass(test_max_difference=__a	)


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

        self._test_inference_batch_single_identical(test_max_difference=__a	)





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







  '''simple docstring'''


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

  def    __SCREAMING_SNAKE_CASE				(		self   :       int	)		->				Any:
        _UpperCamelCase      :	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"	)

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

        _UpperCamelCase      :	List[str]						=       torch.Generator(device="cpu"	).manual_seed(0	)
        _UpperCamelCase      :	Optional[Any]						=       pipe("anime turle"					,   generator=__a					,   output_type="np"	)

        _UpperCamelCase      :	Dict						=       output.images[0]

        assert image.shape == (768, 768, 3)

        assert_mean_pixel_difference(__a					,   __a	)


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

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

        _UpperCamelCase      :	str						=       pipe(
            "anime turtle"					,   prior_num_inference_steps=2					,   num_inference_steps=2					,   output_type="np"					,   )

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







"""simple docstring"""


from collections.abc import Iterator, MutableMapping
from dataclasses import dataclass
from typing import Generic, TypeVar

lowerCamelCase__				  =   TypeVar("KEY")
lowerCamelCase__				  =   TypeVar("VAL")



@dataclass(frozen=_UpperCamelCase		,  slots=_UpperCamelCase   )
class      __SCREAMING_SNAKE_CASE    (						Generic[KEY, VAL]   ):







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__							:KEY
  SCREAMING_SNAKE_CASE__							:VAL





class      __SCREAMING_SNAKE_CASE    (						_Item   ):







  '''simple docstring'''


  def __init__(		self   :       List[str]	)		->				None:
        super().__init__(__a					,   __a	)


  def __bool__(		self   :       Dict	)		->				bool:
        return False





lowerCamelCase__				  =   _DeletedItem()



class      __SCREAMING_SNAKE_CASE    (						MutableMapping[KEY, VAL]   ):







  '''simple docstring'''


  def __init__(		self   :       int					,   __a   :       int = 8					,   __a   :       float = 0.75	)		->				None:
        _UpperCamelCase      :	str						=       initial_block_size
        _UpperCamelCase      :	list[_Item | None]						=       [None] * initial_block_size
        assert 0.0 < capacity_factor < 1.0
        _UpperCamelCase      :	List[str]						=       capacity_factor
        _UpperCamelCase      :	Dict						=       0

  def    __SCREAMING_SNAKE_CASE				(		self   :       int					,   __a   :       KEY	)		->				int:
        return hash(__a	) % len(self._buckets	)

  def    __SCREAMING_SNAKE_CASE				(		self   :       List[Any]					,   __a   :       int	)		->				int:
        return (ind + 1) % len(self._buckets	)

  def    __SCREAMING_SNAKE_CASE				(		self   :       Tuple					,   __a   :       int					,   __a   :       KEY					,   __a   :       VAL	)		->				bool:
        _UpperCamelCase      :	List[Any]						=       self._buckets[ind]
        if not stored:
              _UpperCamelCase      :	Tuple						=       _Item(__a					,   __a	)
              self._len += 1
              return True
        elif stored.key == key:
              _UpperCamelCase      :	Union[str, Any]						=       _Item(__a					,   __a	)
              return True
        else:
              return False

  def    __SCREAMING_SNAKE_CASE				(		self   :       List[str]	)		->				bool:
        _UpperCamelCase      :	Any						=       len(self._buckets	) * self._capacity_factor
        return len(self	) >= int(__a	)

  def    __SCREAMING_SNAKE_CASE				(		self   :       str	)		->				bool:
        if len(self._buckets	) <= self._initial_block_size:
              return False
        _UpperCamelCase      :	List[str]						=       len(self._buckets	) * self._capacity_factor / 2
        return len(self	) < limit

  def    __SCREAMING_SNAKE_CASE				(		self   :       Union[str, Any]					,   __a   :       int	)		->				None:
        _UpperCamelCase      :	Any						=       self._buckets
        _UpperCamelCase      :	List[Any]						=       [None] * new_size
        _UpperCamelCase      :	List[str]						=       0
        for item in old_buckets:
              if item:
                    self._add_item(item.key					,   item.val	)

  def    __SCREAMING_SNAKE_CASE				(		self   :       int	)		->				None:
        self._resize(len(self._buckets	) * 2	)

  def    __SCREAMING_SNAKE_CASE				(		self   :       Union[str, Any]	)		->				None:
        self._resize(len(self._buckets	) // 2	)

  def    __SCREAMING_SNAKE_CASE				(		self   :       List[Any]					,   __a   :       KEY	)		->				Iterator[int]:
        _UpperCamelCase      :	str						=       self._get_bucket_index(__a	)
        for _ in range(len(self._buckets	)	):
              yield ind
              _UpperCamelCase      :	Tuple						=       self._get_next_ind(__a	)

  def    __SCREAMING_SNAKE_CASE				(		self   :       Optional[int]					,   __a   :       KEY					,   __a   :       VAL	)		->				None:
        for ind in self._iterate_buckets(__a	):
              if self._try_set(__a					,   __a					,   __a	):
                    break

  def __setitem__(		self   :       int					,   __a   :       KEY					,   __a   :       VAL	)		->				None:
        if self._is_full():
              self._size_up()

        self._add_item(__a					,   __a	)

  def __delitem__(		self   :       str					,   __a   :       KEY	)		->				None:
        for ind in self._iterate_buckets(__a	):
              _UpperCamelCase      :	Tuple						=       self._buckets[ind]
              if item is None:
                    raise KeyError(__a	)
              if item is _deleted:
                    continue
              if item.key == key:
                    _UpperCamelCase      :	List[Any]						=       _deleted
                    self._len -= 1
                    break
        if self._is_sparse():
              self._size_down()

  def __getitem__(		self   :       str					,   __a   :       KEY	)		->				VAL:
        for ind in self._iterate_buckets(__a	):
              _UpperCamelCase      :	Tuple						=       self._buckets[ind]
              if item is None:
                    break
              if item is _deleted:
                    continue
              if item.key == key:
                    return item.val
        raise KeyError(__a	)

  def __len__(		self   :       List[Any]	)		->				int:
        return self._len

  def __iter__(		self   :       List[str]	)		->				Iterator[KEY]:
        yield from (item.key for item in self._buckets if item)


  def __repr__(		self   :       List[str]	)		->				str:
        _UpperCamelCase      :	Optional[int]						=       " ,".join(
            F'''{item.key}: {item.val}''' for item in self._buckets if item	)
        return F'''HashMap({val_string})'''
