




'''simple docstring'''





import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class        _a    (						__lowerCAmelCase , __lowerCAmelCase , unittest.TestCase							):
     SCREAMING_SNAKE_CASE_							:   List[Any]       =						StableDiffusionXLImgaImgPipeline
     SCREAMING_SNAKE_CASE_							:   List[Any]       =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
     SCREAMING_SNAKE_CASE_							:   Dict       =						PipelineTesterMixin.required_optional_params - {"""latents"""}
     SCREAMING_SNAKE_CASE_							:   Dict       =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     SCREAMING_SNAKE_CASE_							:   Union[str, Any]       =						IMAGE_TO_IMAGE_IMAGE_PARAMS
     SCREAMING_SNAKE_CASE_							:   int       =						IMAGE_TO_IMAGE_IMAGE_PARAMS






     def     _lowercase     (			self   )							->      Optional[Any]:
          torch.manual_seed(0   )
          _snake_case							=				UNetaDConditionModel(
              block_out_channels=(32, 64) ,layers_per_block=2 ,sample_size=32 ,in_channels=4 ,out_channels=4 ,down_block_types=("DownBlock2D", "CrossAttnDownBlock2D") ,up_block_types=("CrossAttnUpBlock2D", "UpBlock2D") ,attention_head_dim=(2, 4) ,use_linear_projection=_SCREAMING_SNAKE_CASE ,addition_embed_type="text_time" ,addition_time_embed_dim=8 ,transformer_layers_per_block=(1, 2) ,projection_class_embeddings_input_dim=80 ,cross_attention_dim=64 ,)
          _snake_case							=				EulerDiscreteScheduler(
              beta_start=0.0_0_0_8_5 ,beta_end=0.0_1_2 ,steps_offset=1 ,beta_schedule="scaled_linear" ,timestep_spacing="leading" ,)
          torch.manual_seed(0   )
          _snake_case							=				AutoencoderKL(
              block_out_channels=[32, 64] ,in_channels=3 ,out_channels=3 ,down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"] ,up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"] ,latent_channels=4 ,sample_size=128 ,)
          torch.manual_seed(0   )
          _snake_case							=				CLIPTextConfig(
              bos_token_id=0 ,eos_token_id=2 ,hidden_size=32 ,intermediate_size=37 ,layer_norm_eps=1e-05 ,num_attention_heads=4 ,num_hidden_layers=5 ,pad_token_id=1 ,vocab_size=1_000 ,hidden_act="gelu" ,projection_dim=32 ,)
          _snake_case							=				CLIPTextModel(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip" ,local_files_only=_SCREAMING_SNAKE_CASE   )

          _snake_case							=				CLIPTextModelWithProjection(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip" ,local_files_only=_SCREAMING_SNAKE_CASE   )

          _snake_case							=				{
              "unet": unet,
              "scheduler": scheduler,
              "vae": vae,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "text_encoder_2": text_encoder_a,
              "tokenizer_2": tokenizer_a,
              # "safety_checker": None,
              # "feature_extractor": None,
          }
          return components






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE=0   )							->      Optional[Any]:
          _snake_case							=				floats_tensor((1, 3, 32, 32) ,rng=random.Random(_SCREAMING_SNAKE_CASE   )   ).to(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				image / 2 + 0.5
          if str(_SCREAMING_SNAKE_CASE   ).startswith("mps"   ):
               _snake_case							=				torch.manual_seed(_SCREAMING_SNAKE_CASE   )
          else:
               _snake_case							=				torch.Generator(device=_SCREAMING_SNAKE_CASE   ).manual_seed(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				{
              "prompt": "A painting of a squirrel eating a burger",
              "image": image,
              "generator": generator,
              "num_inference_steps": 2,
              "guidance_scale": 5.0,
              "output_type": "numpy",
              "strength": 0.7_5,
          }
          return inputs






     def     _lowercase     (			self   )							->      int:
          _snake_case							=				"cpu"  # ensure determinism for the device-dependent torch.Generator
          _snake_case							=				self.get_dummy_components()
          _snake_case							=				StableDiffusionXLImgaImgPipeline(**_SCREAMING_SNAKE_CASE   )
          _snake_case							=				sd_pipe.to(_SCREAMING_SNAKE_CASE   )
          sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE   )

          _snake_case							=				self.get_dummy_inputs(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				sd_pipe(**_SCREAMING_SNAKE_CASE   ).images
          _snake_case							=				image[0, -3:, -3:, -1]

          assert image.shape == (1, 32, 32, 3)

          _snake_case							=				np.array([0.4_6_5_6, 0.4_8_4_0, 0.4_4_3_9, 0.6_6_9_8, 0.5_5_7_4, 0.4_5_2_4, 0.5_7_9_9, 0.5_9_4_3, 0.5_1_6_5]   )

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






     def     _lowercase     (			self   )							->      Union[str, Any]:
          super().test_attention_slicing_forward_pass(expected_max_diff=3e-3   )






     def     _lowercase     (			self   )							->      Dict:
          super().test_inference_batch_single_identical(expected_max_diff=3e-3   )






     def     _lowercase     (			self   )							->      List[Any]:
          pass






     def     _lowercase     (			self   )							->      str:
          _snake_case							=				self.get_dummy_components()
          _snake_case							=				StableDiffusionXLImgaImgPipeline(**_SCREAMING_SNAKE_CASE   )
          _snake_case							=				sd_pipe.to(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				sd_pipe.to(_SCREAMING_SNAKE_CASE   )
          sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE   )

          # forward without prompt embeds
          _snake_case							=				self.get_dummy_inputs(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				3 * ["this is a negative prompt"]
          _snake_case							=				negative_prompt
          _snake_case							=				3 * [inputs["prompt"]]

          _snake_case							=				sd_pipe(**_SCREAMING_SNAKE_CASE   )
          _snake_case							=				output.images[0, -3:, -3:, -1]

          # forward with prompt embeds
          _snake_case							=				self.get_dummy_inputs(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				3 * ["this is a negative prompt"]
          _snake_case							=				3 * [inputs.pop("prompt"   )]

          (
              (
              _snake_case
          )							, (
              _snake_case
          )							, (
              _snake_case
          )							, (
              _snake_case
          )							, 
          )							=				sd_pipe.encode_prompt(_SCREAMING_SNAKE_CASE ,negative_prompt=_SCREAMING_SNAKE_CASE   )

          _snake_case							=				sd_pipe(
              **_SCREAMING_SNAKE_CASE ,prompt_embeds=_SCREAMING_SNAKE_CASE ,negative_prompt_embeds=_SCREAMING_SNAKE_CASE ,pooled_prompt_embeds=_SCREAMING_SNAKE_CASE ,negative_pooled_prompt_embeds=_SCREAMING_SNAKE_CASE ,)
          _snake_case							=				output.images[0, -3:, -3:, -1]

          # make sure that it's equal
          assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()   ).max() < 1e-4







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






     def     _lowercase     (			self   )							->      Any:
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE="cpu" ,_SCREAMING_SNAKE_CASE=torch.floataa ,_SCREAMING_SNAKE_CASE=0   )							->      Dict:
          _snake_case							=				torch.Generator(device=_SCREAMING_SNAKE_CASE   ).manual_seed(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				np.random.RandomState(_SCREAMING_SNAKE_CASE   ).standard_normal((1, 4, 64, 64)   )
          _snake_case							=				torch.from_numpy(_SCREAMING_SNAKE_CASE   ).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE   )
          _snake_case							=				{
              "prompt": "a photograph of an astronaut riding a horse",
              "latents": latents,
              "generator": generator,
              "num_inference_steps": 3,
              "guidance_scale": 7.5,
              "output_type": "numpy",
          }
          return inputs






     def     _lowercase     (			self   )							->      List[str]:
          _snake_case							=				DiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2-base"   )
          pipe.to(_SCREAMING_SNAKE_CASE   )
          pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE   )

          _snake_case							=				self.get_inputs(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				pipe(**_SCREAMING_SNAKE_CASE   ).images
          _snake_case							=				image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)
          _snake_case							=				np.array([0.4_9_4_9_3, 0.4_7_8_9_6, 0.4_0_7_9_8, 0.5_4_2_1_4, 0.5_3_2_1_2, 0.4_8_2_0_2, 0.4_7_6_5_6, 0.4_6_3_2_9, 0.4_8_5_0_6]   )
          assert np.abs(image_slice - expected_slice   ).max() < 7e-3








'''simple docstring'''





from __future__ import annotations

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

UpperCamelCase_						:      int       =						TypeVar('''T''')
UpperCamelCase_						:      Dict       =						TypeVar('''U''')

class        _a    (						Generic[T, U]							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      Dict:
          _snake_case							=				key
          _snake_case							=				val
          _snake_case							=				None
          _snake_case							=				None






     def __repr__(			self   )							->      str:
          return (
              f"""Node: key: {self.key}, val: {self.val}, """
              f"""has next: {bool(self.next   )}, has prev: {bool(self.prev   )}"""
          )







class        _a    (						Generic[T, U]							):






     def __init__(			self   )							->      None:
          _snake_case							=				DoubleLinkedListNode(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          _snake_case							=				DoubleLinkedListNode(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          _snake_case							, _snake_case							=				self.rear, self.head






     def __repr__(			self   )							->      str:
          _snake_case							=				["DoubleLinkedList"]
          _snake_case							=				self.head
          while node.next is not None:
               rep.append(str(_SCREAMING_SNAKE_CASE   )   )
               _snake_case							=				node.next
          rep.append(str(self.rear   )   )
          return ",\n    ".join(_SCREAMING_SNAKE_CASE   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      None:
          _snake_case							=				self.rear.prev

          # All nodes other than self.head are guaranteed to have non-None previous
          assert previous is not None

          _snake_case							=				node
          _snake_case							=				previous
          _snake_case							=				node
          _snake_case							=				self.rear






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      DoubleLinkedListNode[T, U] | None:

          if node.prev is None or node.next is None:
               return None

          _snake_case							=				node.next
          _snake_case							=				node.prev
          _snake_case							=				None
          _snake_case							=				None
          return node







class        _a    (						Generic[T, U]							):
     SCREAMING_SNAKE_CASE_							:   dict[Callable[[T], U], LRUCache[T, U]]       =						{}






     def __init__(			self ,_SCREAMING_SNAKE_CASE   )							->      str:
          _snake_case							=				DoubleLinkedList()
          _snake_case							=				capacity
          _snake_case							=				0
          _snake_case							=				0
          _snake_case							=				0
          _snake_case							=				{}






     def __repr__(			self   )							->      str:
          return (
              f"""CacheInfo(hits={self.hits}, misses={self.miss}, """
              f"""capacity={self.capacity}, current size={self.num_keys})"""
          )






     def __contains__(			self ,_SCREAMING_SNAKE_CASE   )							->      bool:
          return key in self.cache






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      U | None:
          # Note: pythonic interface would throw KeyError rather than return None

          if key in self.cache:
               self.hits += 1
               _snake_case							=				self.cache[key]
               _snake_case							=				self.list.remove(self.cache[key]   )
               assert node == value_node

               # node is guaranteed not None because it is in self.cache
               assert node is not None
               self.list.add(_SCREAMING_SNAKE_CASE   )
               return node.val
          self.miss += 1
          return None






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      None:

          if key not in self.cache:
               if self.num_keys >= self.capacity:
                    # delete first node (oldest) when over capacity
                    _snake_case							=				self.list.head.next

                    # guaranteed to have a non-None first node when num_keys > 0
                    # explain to type checker via assertions
                    assert first_node is not None
                    assert first_node.key is not None
                    assert (
                        self.list.remove(_SCREAMING_SNAKE_CASE   ) is not None
                    )  # node guaranteed to be in list assert node.key is not None

                    del self.cache[first_node.key]
                    self.num_keys -= 1
               _snake_case							=				DoubleLinkedListNode(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
               self.list.add(self.cache[key]   )
               self.num_keys += 1

          else:
               # bump node to the end of the list, update value
               _snake_case							=				self.list.remove(self.cache[key]   )
               assert node is not None  # node guaranteed to be in list
               _snake_case							=				value
               self.list.add(_SCREAMING_SNAKE_CASE   )






     @classmethod
     def     _lowercase     (			cls ,_SCREAMING_SNAKE_CASE = 128   )							->      Callable[[Callable[[T], U]], Callable[..., U]]:

          def cache_decorator_inner(_SCREAMING_SNAKE_CASE   ) -> Callable[..., U]:
               def cache_decorator_wrapper(*_SCREAMING_SNAKE_CASE   ) -> U:
                    if func not in cls.decorator_function_to_instance_map:
                         _snake_case							=				LRUCache(_SCREAMING_SNAKE_CASE   )

                    _snake_case							=				cls.decorator_function_to_instance_map[func].get(args[0]   )
                    if result is None:
                         _snake_case							=				func(*_SCREAMING_SNAKE_CASE   )
                         cls.decorator_function_to_instance_map[func].put(args[0] ,_SCREAMING_SNAKE_CASE   )
                    return result

               def cache_info() -> LRUCache[T, U]:
                    return cls.decorator_function_to_instance_map[func]

               setattr(_SCREAMING_SNAKE_CASE ,"cache_info" ,_SCREAMING_SNAKE_CASE   )  # noqa: B010

               return cache_decorator_wrapper

          return cache_decorator_inner


if __name__ == "__main__":
   import doctest

   doctest.testmod()


