


from __future__ import annotations

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

SCREAMING_SNAKE_CASE__					:     Optional[int]       =     TypeVar('T')
SCREAMING_SNAKE_CASE__					:     List[Any]       =     TypeVar('U')


class 		_SCREAMING_SNAKE_CASE     (						Generic[T, U]     ):


     def __init__(  self							,					A_							,					A_    ):
       _UpperCAmelCase		:							str							=     key
       _UpperCAmelCase		:							Tuple							=     val
       _UpperCAmelCase		:							DoubleLinkedListNode[T, U] | None							=     None
       _UpperCAmelCase		:							DoubleLinkedListNode[T, U] | None							=     None




     def __repr__(  self    ):
       return (
           F'''Node: key: {self.key}, val: {self.val}, '''
           F'''has next: {bool(self.next    )}, has prev: {bool(self.prev    )}'''
       )


class 		_SCREAMING_SNAKE_CASE     (						Generic[T, U]     ):


     def __init__(  self    ):
       _UpperCAmelCase		:							DoubleLinkedListNode[T, U]							=     DoubleLinkedListNode(A_							,					A_    )
       _UpperCAmelCase		:							DoubleLinkedListNode[T, U]							=     DoubleLinkedListNode(A_							,					A_    )
       _UpperCAmelCase,_UpperCAmelCase		:							Any							=     self.rear, self.head


     def __repr__(  self    ):
       _UpperCAmelCase		:							Union[str, Any]							=     ["""DoubleLinkedList"""]
       _UpperCAmelCase		:							List[str]							=     self.head
       while node.next is not None:
         rep.append(str(A_    )    )
         _UpperCAmelCase		:							List[str]							=     node.next
       rep.append(str(self.rear    )    )
       return ",\n    ".join(A_    )


     def        __snake_case(  self							,					A_    ):
       _UpperCAmelCase		:							Optional[int]							=     self.rear.prev

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

       _UpperCAmelCase		:							Any							=     node
       _UpperCAmelCase		:							List[Any]							=     previous
       _UpperCAmelCase		:							Dict							=     node
       _UpperCAmelCase		:							List[Any]							=     self.rear




     def        __snake_case(  self							,					A_    ):

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

       _UpperCAmelCase		:							Dict							=     node.next
       _UpperCAmelCase		:							Tuple							=     node.prev
       _UpperCAmelCase		:							Tuple							=     None
       _UpperCAmelCase		:							Optional[Any]							=     None
       return node


class 		_SCREAMING_SNAKE_CASE     (						Generic[T, U]     ):
     __SCREAMING_SNAKE_CASE     					=    {}


     def __init__(  self							,					A_    ):
       _UpperCAmelCase		:							DoubleLinkedList[T, U]							=     DoubleLinkedList()
       _UpperCAmelCase		:							List[str]							=     capacity
       _UpperCAmelCase		:							Optional[Any]							=     0
       _UpperCAmelCase		:							Any							=     0
       _UpperCAmelCase		:							Union[str, Any]							=     0
       _UpperCAmelCase		:							dict[T, DoubleLinkedListNode[T, U]]							=     {}


     def __repr__(  self    ):
       return (
           F'''CacheInfo(hits={self.hits}, misses={self.miss}, '''
           F'''capacity={self.capacity}, current size={self.num_keys})'''
       )


     def __contains__(  self							,					A_    ):
       return key in self.cache


     def        __snake_case(  self							,					A_    ):
       # Note: pythonic interface would throw KeyError rather than return None

       if key in self.cache:
         self.hits += 1
         _UpperCAmelCase		:							DoubleLinkedListNode[T, U]							=     self.cache[key]
         _UpperCAmelCase		:							List[Any]							=     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(A_    )
         return node.val
       self.miss += 1
       return None


     def        __snake_case(  self							,					A_							,					A_    ):

       if key not in self.cache:
         if self.num_keys >= self.capacity:
           # delete first node (oldest) when over capacity
           _UpperCAmelCase		:							Optional[int]							=     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(A_    ) 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
         _UpperCAmelCase		:							List[str]							=     DoubleLinkedListNode(A_							,					A_    )
         self.list.add(self.cache[key]    )
         self.num_keys += 1

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




     @classmethod
     def        __snake_case(  cls							,					A_ = 1_28    ):

       def cache_decorator_inner(A_    ) -> Callable[..., U]:
         def cache_decorator_wrapper(*A_    ) -> U:
           if func not in cls.decorator_function_to_instance_map:
             _UpperCAmelCase		:							Optional[int]							=     LRUCache(A_    )

           _UpperCAmelCase		:							List[str]							=     cls.decorator_function_to_instance_map[func].get(args[0]    )
           if result is None:
             _UpperCAmelCase		:							str							=     func(*A_    )
             cls.decorator_function_to_instance_map[func].put(args[0]							,					A_    )
           return result

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

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

         return cache_decorator_wrapper

       return cache_decorator_inner


if __name__ == "__main__":
 import doctest

 doctest.testmod()









import gc
import random
import unittest

import torch

from diffusers import (
    IFImgaImgPipeline,
    IFImgaImgSuperResolutionPipeline,
    IFInpaintingPipeline,
    IFInpaintingSuperResolutionPipeline,
    IFPipeline,
    IFSuperResolutionPipeline,
)
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import floats_tensor, load_numpy, require_torch_gpu, skip_mps, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference
from . import IFPipelineTesterMixin


@skip_mps
class 		_SCREAMING_SNAKE_CASE     (						A							,	A							,	unittest.TestCase     ):
     __SCREAMING_SNAKE_CASE     					=    IFPipeline
     __SCREAMING_SNAKE_CASE     					=    TEXT_TO_IMAGE_PARAMS - {'''width''', '''height''', '''latents'''}
     __SCREAMING_SNAKE_CASE     					=    TEXT_TO_IMAGE_BATCH_PARAMS
     __SCREAMING_SNAKE_CASE     					=    PipelineTesterMixin.required_optional_params - {'''latents'''}


     def        __snake_case(  self    ):
       return self._get_dummy_components()


     def        __snake_case(  self							,					A_							,					A_=0    ):
       if str(A_    ).startswith("""mps"""    ):
         _UpperCAmelCase		:							Tuple							=     torch.manual_seed(A_    )
       else:
         _UpperCAmelCase		:							Optional[Any]							=     torch.Generator(device=A_    ).manual_seed(A_    )

       _UpperCAmelCase		:							Any							=     {
           """prompt""": """A painting of a squirrel eating a burger""",
           """generator""": generator,
           """num_inference_steps""": 2,
           """output_type""": """numpy""",
       }

       return inputs


     def        __snake_case(  self    ):
       self._test_save_load_optional_components()


     @unittest.skipIf(torch_device != """cuda"""							,					reason="""float16 requires CUDA"""    )
     def        __snake_case(  self    ):
       # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
       super().test_save_load_floataa(expected_max_diff=1e-1    )


     def        __snake_case(  self    ):
       self._test_attention_slicing_forward_pass(expected_max_diff=1e-2    )


     def        __snake_case(  self    ):
       self._test_save_load_local()


     def        __snake_case(  self    ):
       self._test_inference_batch_single_identical(
           expected_max_diff=1e-2							,					)




     @unittest.skipIf(
         torch_device != """cuda""" or not is_xformers_available()							,					reason="""XFormers attention is only available with CUDA and `xformers` installed"""							,					)
     def        __snake_case(  self    ):
       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3    )


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


     def        __snake_case(  self    ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()


     def        __snake_case(  self    ):
       # if

       _UpperCAmelCase		:							List[str]							=     IFPipeline.from_pretrained("""DeepFloyd/IF-I-XL-v1.0"""							,					variant="""fp16"""							,					torch_dtype=torch.floataa    )

       _UpperCAmelCase		:							Dict							=     IFSuperResolutionPipeline.from_pretrained(
           """DeepFloyd/IF-II-L-v1.0"""							,					variant="""fp16"""							,					torch_dtype=torch.floataa							,					text_encoder=A_							,					tokenizer=A_    )

       # pre compute text embeddings and remove T5 to save memory

       pipe_a.text_encoder.to("""cuda"""    )

       _UpperCAmelCase,_UpperCAmelCase		:							Dict							=     pipe_a.encode_prompt("""anime turtle"""							,					device="""cuda"""    )

       del pipe_a.tokenizer
       del pipe_a.text_encoder
       gc.collect()

       _UpperCAmelCase		:							List[Any]							=     None
       _UpperCAmelCase		:							Any							=     None

       pipe_a.enable_model_cpu_offload()
       pipe_a.enable_model_cpu_offload()

       pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )
       pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )

       self._test_if(A_							,					A_							,					A_							,					A_    )

       pipe_a.remove_all_hooks()
       pipe_a.remove_all_hooks()

       # img2img

       _UpperCAmelCase		:							Any							=     IFImgaImgPipeline(**pipe_a.components    )
       _UpperCAmelCase		:							Union[str, Any]							=     IFImgaImgSuperResolutionPipeline(**pipe_a.components    )

       pipe_a.enable_model_cpu_offload()
       pipe_a.enable_model_cpu_offload()

       pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )
       pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )

       self._test_if_imgaimg(A_							,					A_							,					A_							,					A_    )

       pipe_a.remove_all_hooks()
       pipe_a.remove_all_hooks()

       # inpainting

       _UpperCAmelCase		:							Optional[int]							=     IFInpaintingPipeline(**pipe_a.components    )
       _UpperCAmelCase		:							Union[str, Any]							=     IFInpaintingSuperResolutionPipeline(**pipe_a.components    )

       pipe_a.enable_model_cpu_offload()
       pipe_a.enable_model_cpu_offload()

       pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )
       pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )

       self._test_if_inpainting(A_							,					A_							,					A_							,					A_    )


     def        __snake_case(  self							,					A_							,					A_							,					A_							,					A_    ):
       # pipeline 1

       _start_torch_memory_measurement()

       _UpperCAmelCase		:							Optional[Any]							=     torch.Generator(device="""cpu"""    ).manual_seed(0    )
       _UpperCAmelCase		:							List[str]							=     pipe_a(
           prompt_embeds=A_							,					negative_prompt_embeds=A_							,					num_inference_steps=2							,					generator=A_							,					output_type="""np"""							,					)

       _UpperCAmelCase		:							int							=     output.images[0]

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

       _UpperCAmelCase		:							Dict							=     torch.cuda.max_memory_allocated()
       assert mem_bytes < 13 * 10**9

       _UpperCAmelCase		:							str							=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if.npy"""    )
       assert_mean_pixel_difference(A_							,					A_    )

       # pipeline 2

       _start_torch_memory_measurement()

       _UpperCAmelCase		:							List[Any]							=     torch.Generator(device="""cpu"""    ).manual_seed(0    )

       _UpperCAmelCase		:							str							=     floats_tensor((1, 3, 64, 64)							,					rng=random.Random(0    )    ).to(A_    )

       _UpperCAmelCase		:							str							=     pipe_a(
           prompt_embeds=A_							,					negative_prompt_embeds=A_							,					image=A_							,					generator=A_							,					num_inference_steps=2							,					output_type="""np"""							,					)

       _UpperCAmelCase		:							int							=     output.images[0]

       assert image.shape == (2_56, 2_56, 3)

       _UpperCAmelCase		:							Any							=     torch.cuda.max_memory_allocated()
       assert mem_bytes < 4 * 10**9

       _UpperCAmelCase		:							List[Any]							=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_superresolution_stage_II.npy"""    )
       assert_mean_pixel_difference(A_							,					A_    )


     def        __snake_case(  self							,					A_							,					A_							,					A_							,					A_    ):
       # pipeline 1

       _start_torch_memory_measurement()

       _UpperCAmelCase		:							Optional[int]							=     floats_tensor((1, 3, 64, 64)							,					rng=random.Random(0    )    ).to(A_    )

       _UpperCAmelCase		:							Dict							=     torch.Generator(device="""cpu"""    ).manual_seed(0    )

       _UpperCAmelCase		:							List[str]							=     pipe_a(
           prompt_embeds=A_							,					negative_prompt_embeds=A_							,					image=A_							,					num_inference_steps=2							,					generator=A_							,					output_type="""np"""							,					)

       _UpperCAmelCase		:							int							=     output.images[0]

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

       _UpperCAmelCase		:							str							=     torch.cuda.max_memory_allocated()
       assert mem_bytes < 10 * 10**9

       _UpperCAmelCase		:							str							=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img.npy"""    )
       assert_mean_pixel_difference(A_							,					A_    )

       # pipeline 2

       _start_torch_memory_measurement()

       _UpperCAmelCase		:							List[str]							=     torch.Generator(device="""cpu"""    ).manual_seed(0    )

       _UpperCAmelCase		:							Union[str, Any]							=     floats_tensor((1, 3, 2_56, 2_56)							,					rng=random.Random(0    )    ).to(A_    )
       _UpperCAmelCase		:							Optional[int]							=     floats_tensor((1, 3, 64, 64)							,					rng=random.Random(0    )    ).to(A_    )

       _UpperCAmelCase		:							Optional[int]							=     pipe_a(
           prompt_embeds=A_							,					negative_prompt_embeds=A_							,					image=A_							,					original_image=A_							,					generator=A_							,					num_inference_steps=2							,					output_type="""np"""							,					)

       _UpperCAmelCase		:							Dict							=     output.images[0]

       assert image.shape == (2_56, 2_56, 3)

       _UpperCAmelCase		:							Any							=     torch.cuda.max_memory_allocated()
       assert mem_bytes < 4 * 10**9

       _UpperCAmelCase		:							int							=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img_superresolution_stage_II.npy"""    )
       assert_mean_pixel_difference(A_							,					A_    )




     def        __snake_case(  self							,					A_							,					A_							,					A_							,					A_    ):
       # pipeline 1

       _start_torch_memory_measurement()

       _UpperCAmelCase		:							List[Any]							=     floats_tensor((1, 3, 64, 64)							,					rng=random.Random(0    )    ).to(A_    )
       _UpperCAmelCase		:							Dict							=     floats_tensor((1, 3, 64, 64)							,					rng=random.Random(1    )    ).to(A_    )

       _UpperCAmelCase		:							Optional[Any]							=     torch.Generator(device="""cpu"""    ).manual_seed(0    )
       _UpperCAmelCase		:							str							=     pipe_a(
           prompt_embeds=A_							,					negative_prompt_embeds=A_							,					image=A_							,					mask_image=A_							,					num_inference_steps=2							,					generator=A_							,					output_type="""np"""							,					)

       _UpperCAmelCase		:							Optional[Any]							=     output.images[0]

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

       _UpperCAmelCase		:							List[str]							=     torch.cuda.max_memory_allocated()
       assert mem_bytes < 10 * 10**9

       _UpperCAmelCase		:							Dict							=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting.npy"""    )
       assert_mean_pixel_difference(A_							,					A_    )

       # pipeline 2

       _start_torch_memory_measurement()

       _UpperCAmelCase		:							str							=     torch.Generator(device="""cpu"""    ).manual_seed(0    )

       _UpperCAmelCase		:							Optional[int]							=     floats_tensor((1, 3, 64, 64)							,					rng=random.Random(0    )    ).to(A_    )
       _UpperCAmelCase		:							Tuple							=     floats_tensor((1, 3, 2_56, 2_56)							,					rng=random.Random(0    )    ).to(A_    )
       _UpperCAmelCase		:							str							=     floats_tensor((1, 3, 2_56, 2_56)							,					rng=random.Random(1    )    ).to(A_    )

       _UpperCAmelCase		:							Union[str, Any]							=     pipe_a(
           prompt_embeds=A_							,					negative_prompt_embeds=A_							,					image=A_							,					mask_image=A_							,					original_image=A_							,					generator=A_							,					num_inference_steps=2							,					output_type="""np"""							,					)

       _UpperCAmelCase		:							Any							=     output.images[0]

       assert image.shape == (2_56, 2_56, 3)

       _UpperCAmelCase		:							Optional[int]							=     torch.cuda.max_memory_allocated()
       assert mem_bytes < 4 * 10**9

       _UpperCAmelCase		:							Tuple							=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting_superresolution_stage_II.npy"""    )
       assert_mean_pixel_difference(A_							,					A_    )




def    a__    (		):
  torch.cuda.empty_cache()
  torch.cuda.reset_max_memory_allocated()
  torch.cuda.reset_peak_memory_stats()





