


'''simple docstring'''







import gc
import unittest

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

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

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()



class 				__lowerCamelCase				( A__       ,      A__       ,      unittest.TestCase ):
    """simple docstring"""
    a									=					StableDiffusionXLPipeline
    a									=					TEXT_TO_IMAGE_PARAMS
    a									=					TEXT_TO_IMAGE_BATCH_PARAMS
    a									=					TEXT_TO_IMAGE_IMAGE_PARAMS
    a									=					TEXT_TO_IMAGE_IMAGE_PARAMS


    def      A					(						self    :							Dict):
           torch.manual_seed(0)
           _A  :			List[str]											=      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=_lowerCAmelCase   ,							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   ,							)
           _A  :			Tuple											=      EulerDiscreteScheduler(
               beta_start=0.0_0085   ,							beta_end=0.012   ,							steps_offset=1   ,							beta_schedule='scaled_linear'   ,							timestep_spacing='leading'   ,							)
           torch.manual_seed(0)
           _A  :			int											=      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)
           _A  :			List[Any]											=      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=1000   ,							hidden_act='gelu'   ,							projection_dim=32   ,							)
           _A  :			Tuple											=      CLIPTextModel(_lowerCAmelCase)
           _A  :			str											=      CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   ,							local_files_only=_lowerCAmelCase)

           _A  :			str											=      CLIPTextModelWithProjection(_lowerCAmelCase)
           _A  :			str											=      CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   ,							local_files_only=_lowerCAmelCase)

           _A  :			Union[str, Any]											=      {
               '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      A					(						self    :							Optional[Any]   ,							SCREAMING_SNAKE_CASE    :							int   ,							SCREAMING_SNAKE_CASE    :							List[Any]=0):
           if str(_lowerCAmelCase).startswith('mps'):
                  _A  :			Dict											=      torch.manual_seed(_lowerCAmelCase)
           else:
                  _A  :			Optional[Any]											=      torch.Generator(device=_lowerCAmelCase).manual_seed(_lowerCAmelCase)
           _A  :			Union[str, Any]											=      {
               'prompt': 'A painting of a squirrel eating a burger',
               'generator': generator,
               'num_inference_steps': 2,
               'guidance_scale': 5.0,
               'output_type': 'numpy',
           }
           return inputs


    def      A					(						self    :							Optional[Any]):
           _A  :			Dict											=      'cpu'  # ensure determinism for the device-dependent torch.Generator
           _A  :			Union[str, Any]											=      self.get_dummy_components()
           _A  :			Optional[Any]											=      StableDiffusionXLPipeline(**_lowerCAmelCase)
           _A  :			Tuple											=      sd_pipe.to(_lowerCAmelCase)
           sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase)

           _A  :			Optional[int]											=      self.get_dummy_inputs(_lowerCAmelCase)
           _A  :			int											=      sd_pipe(**_lowerCAmelCase).images
           _A  :			Optional[int]											=      image[0, -3:, -3:, -1]

           assert image.shape == (1, 64, 64, 3)
           _A  :			int											=      np.array([0.5873, 0.6128, 0.4797, 0.5122, 0.5674, 0.4639, 0.5227, 0.5149, 0.4747])

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


    def      A					(						self    :							List[Any]):
           _A  :			Optional[int]											=      self.get_dummy_components()
           _A  :			List[str]											=      StableDiffusionXLPipeline(**_lowerCAmelCase)
           _A  :			Union[str, Any]											=      sd_pipe.to(_lowerCAmelCase)
           _A  :			List[Any]											=      sd_pipe.to(_lowerCAmelCase)
           sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase)

           # forward without prompt embeds
           _A  :			Dict											=      self.get_dummy_inputs(_lowerCAmelCase)
           _A  :			Tuple											=      3 * ['this is a negative prompt']
           _A  :			int											=      negative_prompt
           _A  :			int											=      3 * [inputs['prompt']]

           _A  :			Tuple											=      sd_pipe(**_lowerCAmelCase)
           _A  :			Optional[Any]											=      output.images[0, -3:, -3:, -1]

           # forward with prompt embeds
           _A  :			str											=      self.get_dummy_inputs(_lowerCAmelCase)
           _A  :			int											=      3 * ['this is a negative prompt']
           _A  :			Optional[int]											=      3 * [inputs.pop('prompt')]

           (
               (
               _A    
           )						,				(
               _A    
           )						,				(
               _A    
           )						,				(
               _A    
           )						,				    
           )  :			str											=      sd_pipe.encode_prompt(_lowerCAmelCase   ,							negative_prompt=_lowerCAmelCase)

           _A  :			Dict											=      sd_pipe(
               **_lowerCAmelCase   ,							prompt_embeds=_lowerCAmelCase   ,							negative_prompt_embeds=_lowerCAmelCase   ,							pooled_prompt_embeds=_lowerCAmelCase   ,							negative_pooled_prompt_embeds=_lowerCAmelCase   ,							)
           _A  :			Any											=      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


    def      A					(						self    :							List[Any]):
           super().test_attention_slicing_forward_pass(expected_max_diff=3e-3)


    def      A					(						self    :							Tuple):
           super().test_inference_batch_single_identical(expected_max_diff=3e-3)




@slow
@require_torch_gpu
class 				__lowerCamelCase				( unittest.TestCase ):
    """simple docstring"""


    def      A					(						self    :							str):
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()


    def      A					(						self    :							Optional[int]   ,							SCREAMING_SNAKE_CASE    :							Dict   ,							SCREAMING_SNAKE_CASE    :							List[str]="cpu"   ,							SCREAMING_SNAKE_CASE    :							Union[str, Any]=torch.floataa   ,							SCREAMING_SNAKE_CASE    :							Optional[int]=0):
           _A  :			str											=      torch.Generator(device=_lowerCAmelCase).manual_seed(_lowerCAmelCase)
           _A  :			List[Any]											=      np.random.RandomState(_lowerCAmelCase).standard_normal((1, 4, 64, 64))
           _A  :			List[Any]											=      torch.from_numpy(_lowerCAmelCase).to(device=_lowerCAmelCase   ,							dtype=_lowerCAmelCase)
           _A  :			int											=      {
               '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      A					(						self    :							Dict):
           _A  :			Optional[int]											=      DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base')
           pipe.to(_lowerCAmelCase)
           pipe.set_progress_bar_config(disable=_lowerCAmelCase)

           _A  :			str											=      self.get_inputs(_lowerCAmelCase)
           _A  :			Any											=      pipe(**_lowerCAmelCase).images
           _A  :			List[str]											=      image[0, -3:, -3:, -1].flatten()

           assert image.shape == (1, 512, 512, 3)
           _A  :			Optional[int]											=      np.array([0.4_9493, 0.4_7896, 0.4_0798, 0.5_4214, 0.5_3212, 0.4_8202, 0.4_7656, 0.4_6329, 0.4_8506])
           assert np.abs(image_slice - expected_slice).max() < 7e-3










'''simple docstring'''







class 				__lowerCamelCase				:
    """simple docstring"""


    def __init__(						self    :							Optional[Any]):
           _A  :			Optional[Any]											=      [0]
           _A  :			Optional[Any]											=      0


    def      A					(						self    :							List[Any]   ,							SCREAMING_SNAKE_CASE    :							int):
           _A  :			Optional[Any]											=      self.__heap[i]
           while i // 2 > 0:
                  if self.__heap[i] > self.__heap[i // 2]:
                         _A  :			Dict											=      self.__heap[i // 2]
                         _A  :			Optional[Any]											=      temporary
                  i //= 2


    def      A					(						self    :							List[Any]   ,							SCREAMING_SNAKE_CASE    :							int):
           self.__heap.append(SCREAMING_SNAKE_CASE)
           self.__size += 1
           self.__swap_up(self.__size)


    def      A					(						self    :							int   ,							SCREAMING_SNAKE_CASE    :							int):
           while self.__size >= 2 * i:
                  if 2 * i + 1 > self.__size:
                         _A  :			Optional[int]											=      2 * i
                  else:
                         if self.__heap[2 * i] > self.__heap[2 * i + 1]:
                                _A  :			Any											=      2 * i
                         else:
                                _A  :			Optional[Any]											=      2 * i + 1
                  _A  :			List[str]											=      self.__heap[i]
                  if self.__heap[i] < self.__heap[bigger_child]:
                         _A  :			Union[str, Any]											=      self.__heap[bigger_child]
                         _A  :			Optional[int]											=      temporary
                  _A  :			Any											=      bigger_child


    def      A					(						self    :							List[str]):
           _A  :			str											=      self.__heap[1]
           _A  :			Tuple											=      self.__heap[self.__size]
           self.__size -= 1
           self.__heap.pop()
           self.__swap_down(1)
           return max_value


    @property
    def      A					(						self    :							int):
           return self.__heap[1:]


    def __len__(						self    :							List[str]):
           return self.__size


if __name__ == "__main__":
   import doctest

   doctest.testmod()
   # create an instance of BinaryHeap
   A   :  Dict       		=				BinaryHeap()
   binary_heap.insert(6)
   binary_heap.insert(10)
   binary_heap.insert(15)
   binary_heap.insert(12)
   # pop root(max-values because it is max heap)
   print(binary_heap.pop())  # 15
   print(binary_heap.pop())  # 12
   # get the list and size after operations
   print(binary_heap.get_list)
   print(len(binary_heap))






