



'''simple docstring'''





import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    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						):
    __UpperCAmelCase							=	StableDiffusionParadigmsPipeline
    __UpperCAmelCase							=	TEXT_TO_IMAGE_PARAMS
    __UpperCAmelCase							=	TEXT_TO_IMAGE_BATCH_PARAMS
    __UpperCAmelCase							=	TEXT_TO_IMAGE_IMAGE_PARAMS
    __UpperCAmelCase							=	TEXT_TO_IMAGE_IMAGE_PARAMS





    def 					__snake_case       (    self    :    Union[str, Any]     ):



       '''simple docstring'''







       torch.manual_seed(0     )
       snake_case   :							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'''),    cross_attention_dim=32,    attention_head_dim=(2, 4),    use_linear_projection=_snake_case,    )
       snake_case   :							Union[str, Any]						  =DDIMParallelScheduler(
           beta_start=0.0_0085,    beta_end=0.012,    beta_schedule='''scaled_linear''',    clip_sample=_snake_case,    set_alpha_to_one=_snake_case,    )
       torch.manual_seed(0     )
       snake_case   :							List[str]						  =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   :							Union[str, 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=1_000,    hidden_act='''gelu''',    projection_dim=512,    )
       snake_case   :							int						  =CLIPTextModel(_snake_case     )
       snake_case   :							str						  =CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''     )

       snake_case   :							str						  ={
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''safety_checker''': None,
           '''feature_extractor''': None,
       }
       return components





    def 					__snake_case       (    self    :    str,    _snake_case    :    Any,    _snake_case    :    str=0     ):



       '''simple docstring'''







       if str(_snake_case     ).startswith('''mps'''     ):
          snake_case   :							Dict						  =torch.manual_seed(_snake_case     )
       else:
          snake_case   :							Dict						  =torch.Generator(device=_snake_case     ).manual_seed(_snake_case     )
       snake_case   :							List[Any]						  ={
           '''prompt''': '''a photograph of an astronaut riding a horse''',
           '''generator''': generator,
           '''num_inference_steps''': 10,
           '''guidance_scale''': 6.0,
           '''output_type''': '''numpy''',
           '''parallel''': 3,
           '''debug''': True,
       }
       return inputs





    def 					__snake_case       (    self    :    Union[str, Any]     ):



       '''simple docstring'''







       snake_case   :							Optional[Any]						  ='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       snake_case   :							Tuple						  =self.get_dummy_components()
       snake_case   :							int						  =StableDiffusionParadigmsPipeline(**_snake_case     )
       snake_case   :							Tuple						  =sd_pipe.to(_snake_case     )
       sd_pipe.set_progress_bar_config(disable=_snake_case     )

       snake_case   :							Tuple						  =self.get_dummy_inputs(_snake_case     )
       snake_case   :							Optional[int]						  =sd_pipe(**_snake_case     ).images
       snake_case   :							List[Any]						  =image[0, -3:, -3:, -1]
       assert image.shape == (1, 64, 64, 3)

       snake_case   :							Any						  =np.array([0.4773, 0.5417, 0.4723, 0.4925, 0.5631, 0.4752, 0.5240, 0.4935, 0.5023]     )

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





    def 					__snake_case       (    self    :    List[Any]     ):



       '''simple docstring'''







       snake_case   :							Union[str, Any]						  ='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       snake_case   :							Dict						  =self.get_dummy_components()
       torch.manual_seed(0     )
       snake_case   :							List[str]						  =DDPMParallelScheduler()
       torch.manual_seed(0     )
       snake_case   :							str						  =StableDiffusionParadigmsPipeline(**_snake_case     )
       snake_case   :							List[Any]						  =sd_pipe.to(_snake_case     )
       sd_pipe.set_progress_bar_config(disable=_snake_case     )

       snake_case   :							Union[str, Any]						  =self.get_dummy_inputs(_snake_case     )
       snake_case   :							Dict						  =sd_pipe(**_snake_case     ).images
       snake_case   :							Optional[Any]						  =image[0, -3:, -3:, -1]
       assert image.shape == (1, 64, 64, 3)

       snake_case   :							Optional[int]						  =np.array([0.3573, 0.4420, 0.4960, 0.4799, 0.3796, 0.3879, 0.4819, 0.4365, 0.4468]     )

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





    def 					__snake_case       (    self    :    Optional[Any]     ):



       '''simple docstring'''







       super().test_inference_batch_consistent(batch_sizes=[1, 2]     )





    def 					__snake_case       (    self    :    Union[str, Any]     ):



       '''simple docstring'''







       super().test_inference_batch_single_identical(batch_size=2,    expected_max_diff=3E-3     )






    def 					__snake_case       (    self    :    int     ):



       '''simple docstring'''







       snake_case   :							Optional[int]						  ='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       snake_case   :							Tuple						  =self.get_dummy_components()
       snake_case   :							Union[str, Any]						  =StableDiffusionParadigmsPipeline(**_snake_case     )
       snake_case   :							Union[str, Any]						  =sd_pipe.to(_snake_case     )
       sd_pipe.set_progress_bar_config(disable=_snake_case     )

       snake_case   :							str						  =self.get_dummy_inputs(_snake_case     )
       snake_case   :							List[str]						  ='''french fries'''
       snake_case   :							int						  =sd_pipe(**_snake_case,    negative_prompt=_snake_case     )
       snake_case   :							Any						  =output.images
       snake_case   :							List[Any]						  =image[0, -3:, -3:, -1]

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

       snake_case   :							Tuple						  =np.array([0.4771, 0.5420, 0.4683, 0.4918, 0.5636, 0.4725, 0.5230, 0.4923, 0.5015]     )

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





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





    def 					__snake_case       (    self    :    Dict     ):



       '''simple docstring'''







       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()





    def 					__snake_case       (    self    :    List[str],    _snake_case    :    Any=0     ):



       '''simple docstring'''







       snake_case   :							Union[str, Any]						  =torch.Generator(device=_snake_case     ).manual_seed(_snake_case     )
       snake_case   :							Optional[Any]						  ={
           '''prompt''': '''a photograph of an astronaut riding a horse''',
           '''generator''': generator,
           '''num_inference_steps''': 10,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
           '''parallel''': 3,
           '''debug''': True,
       }
       return inputs






    def 					__snake_case       (    self    :    int     ):



       '''simple docstring'''







       snake_case   :							Any						  ='''stabilityai/stable-diffusion-2-base'''
       snake_case   :							Dict						  =DDIMParallelScheduler.from_pretrained(_snake_case,    subfolder='''scheduler'''     )
       snake_case   :							Tuple						  =StableDiffusionParadigmsPipeline.from_pretrained(_snake_case,    scheduler=_snake_case,    safety_checker=_snake_case     )
       pipe.to(_snake_case     )
       pipe.set_progress_bar_config(disable=_snake_case     )
       pipe.enable_attention_slicing()

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

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

       snake_case   :							List[str]						  =np.array([0.9622, 0.9602, 0.9748, 0.9591, 0.9630, 0.9691, 0.9661, 0.9631, 0.9741]     )

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










'''simple docstring'''





class 							lowerCAmelCase_		:





    def __init__(    self    :    List[Any],    _snake_case    :    Any     ):



       '''simple docstring'''







       snake_case   :							Tuple						  =data
       snake_case   :							int						  =None
       snake_case   :							List[Any]						  =None






    def __str__(    self    :    List[str]     ):



       '''simple docstring'''







       return f'''{self.data}'''





class 							lowerCAmelCase_		:





    def __init__(    self    :    Optional[int]     ):



       '''simple docstring'''







       snake_case   :							Optional[int]						  =None
       snake_case   :							int						  =None





    def __iter__(    self    :    List[str]     ):



       '''simple docstring'''







       snake_case   :							str						  =self.head
       while node:
          yield node.data
          snake_case   :							Union[str, Any]						  =node.next





    def __str__(    self    :    List[str]     ):



       '''simple docstring'''







       return "->".join([str(_snake_case     ) for item in self]     )





    def __len__(    self    :    Dict     ):



       '''simple docstring'''







       return sum(1 for _ in self     )





    def 					__snake_case       (    self    :    List[Any],    _snake_case    :    str     ):



       '''simple docstring'''







       self.insert_at_nth(0,    _snake_case     )





    def 					__snake_case       (    self    :    Optional[Any],    _snake_case    :    Dict     ):



       '''simple docstring'''







       self.insert_at_nth(len(self     ),    _snake_case     )





    def 					__snake_case       (    self    :    Union[str, Any],    _snake_case    :    int,    _snake_case    :    Dict     ):



       '''simple docstring'''







       snake_case   :							Optional[int]						  =len(self     )

       if not 0 <= index <= length:
          raise IndexError('''list index out of range'''     )
       snake_case   :							int						  =Node(_snake_case     )
       if self.head is None:
          snake_case   :							Optional[Any]						  =new_node
       elif index == 0:
          snake_case   :							int						  =new_node
          snake_case   :							Union[str, Any]						  =self.head
          snake_case   :							Optional[int]						  =new_node
       elif index == length:
          snake_case   :							Dict						  =new_node
          snake_case   :							int						  =self.tail
          snake_case   :							Union[str, Any]						  =new_node
       else:
          snake_case   :							Optional[int]						  =self.head
          for _ in range(0,    _snake_case     ):
             snake_case   :							Union[str, Any]						  =temp.next
          snake_case   :							int						  =new_node
          snake_case   :							Union[str, Any]						  =temp.previous
          snake_case   :							List[Any]						  =temp
          snake_case   :							int						  =new_node





    def 					__snake_case       (    self    :    str     ):



       '''simple docstring'''







       return self.delete_at_nth(0     )





    def 					__snake_case       (    self    :    List[Any]     ):



       '''simple docstring'''







       return self.delete_at_nth(len(self     ) - 1     )





    def 					__snake_case       (    self    :    Union[str, Any],    _snake_case    :    int     ):



       '''simple docstring'''







       snake_case   :							Dict						  =len(self     )

       if not 0 <= index <= length - 1:
          raise IndexError('''list index out of range'''     )
       snake_case   :							List[str]						  =self.head  # default first node
       if length == 1:
          snake_case   :							Union[str, Any]						  =None
       elif index == 0:
          snake_case   :							Tuple						  =self.head.next
          snake_case   :							Optional[int]						  =None
       elif index == length - 1:
          snake_case   :							Union[str, Any]						  =self.tail
          snake_case   :							List[str]						  =self.tail.previous
          snake_case   :							int						  =None
       else:
          snake_case   :							int						  =self.head
          for _ in range(0,    _snake_case     ):
             snake_case   :							Any						  =temp.next
          snake_case   :							List[Any]						  =temp
          snake_case   :							Optional[Any]						  =temp.previous
          snake_case   :							int						  =temp.next
       return delete_node.data





    def 					__snake_case       (    self    :    List[Any],    _snake_case    :    Tuple     ):



       '''simple docstring'''







       snake_case   :							List[Any]						  =self.head

       while current.data != data:  # Find the position to delete
          if current.next:
             snake_case   :							Any						  =current.next
          else:  # We have reached the end an no value matches
             raise ValueError('''No data matching given value'''     )

       if current == self.head:
          self.delete_head()

       elif current == self.tail:
          self.delete_tail()

       else:  # Before: 1 <--> 2(current) <--> 3
          snake_case   :							Optional[int]						  =current.next  # 1 --> 3
          snake_case   :							List[Any]						  =current.previous  # 1 <--> 3
       return data






    def 					__snake_case       (    self    :    Any     ):



       '''simple docstring'''







       return len(self     ) == 0




def 		_a   (  ):
   snake_case   :							Optional[Any]						  =DoublyLinkedList()
   assert linked_list.is_empty() is True
   assert str(lowerCamelCase_							) == ""

   try:
      linked_list.delete_head()
      raise AssertionError  # This should not happen.
   except IndexError:
      assert True  # This should happen.

   try:
      linked_list.delete_tail()
      raise AssertionError  # This should not happen.
   except IndexError:
      assert True  # This should happen.

   for i in range(10							):
      assert len(lowerCamelCase_							) == i
      linked_list.insert_at_nth(lowerCamelCase_					,						i + 1							)
   assert str(lowerCamelCase_							) == "->".join(str(lowerCamelCase_							) for i in range(1					,						11							)							)

   linked_list.insert_at_head(0							)
   linked_list.insert_at_tail(11							)
   assert str(lowerCamelCase_							) == "->".join(str(lowerCamelCase_							) for i in range(0					,						12							)							)

   assert linked_list.delete_head() == 0
   assert linked_list.delete_at_nth(9							) == 10
   assert linked_list.delete_tail() == 11
   assert len(lowerCamelCase_							) == 9
   assert str(lowerCamelCase_							) == "->".join(str(lowerCamelCase_							) for i in range(1					,						10							)							)


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

      testmod()





