

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    A_				( __lowerCAmelCase	,       __lowerCAmelCase	,       unittest.TestCase  ):


    '''simple docstring'''






    a__       							=    StableDiffusionParadigmsPipeline
    a__       							=    TEXT_TO_IMAGE_PARAMS
    a__       							=    TEXT_TO_IMAGE_BATCH_PARAMS
    a__       							=    TEXT_TO_IMAGE_IMAGE_PARAMS
    a__       							=    TEXT_TO_IMAGE_IMAGE_PARAMS


    def        lowerCAmelCase_  (self					)       -> Optional[int]:
        torch.manual_seed(0					)
        __UpperCAmelCase			   =     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=__SCREAMING_SNAKE_CASE       , )
        __UpperCAmelCase			   =     DDIMParallelScheduler(
            beta_start=0.00085       , beta_end=0.012       , beta_schedule='''scaled_linear'''       , clip_sample=__SCREAMING_SNAKE_CASE       , set_alpha_to_one=__SCREAMING_SNAKE_CASE       , )
        torch.manual_seed(0					)
        __UpperCAmelCase			   =     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					)
        __UpperCAmelCase			   =     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       , )
        __UpperCAmelCase			   =     CLIPTextModel(__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

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


    def        lowerCAmelCase_  (self       , lowercase__       , lowercase__=0					)       -> Optional[int]:




        if str(__SCREAMING_SNAKE_CASE					).startswith('''mps'''					):
            __UpperCAmelCase			   =     torch.manual_seed(__SCREAMING_SNAKE_CASE					)
        else:
            __UpperCAmelCase			   =     torch.Generator(device=__SCREAMING_SNAKE_CASE					).manual_seed(__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     {
            '''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        lowerCAmelCase_  (self					)       -> Optional[int]:
        __UpperCAmelCase			   =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        __UpperCAmelCase			   =     self.get_dummy_components()
        __UpperCAmelCase			   =     StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     sd_pipe.to(__SCREAMING_SNAKE_CASE					)
        sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE					)

        __UpperCAmelCase			   =     self.get_dummy_inputs(__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     sd_pipe(**__SCREAMING_SNAKE_CASE					).images
        __UpperCAmelCase			   =     image[0, -3:, -3:, -1]
        assert image.shape == (1, 64, 64, 3)

        __UpperCAmelCase			   =     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        lowerCAmelCase_  (self					)       -> int:
        __UpperCAmelCase			   =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        __UpperCAmelCase			   =     self.get_dummy_components()
        torch.manual_seed(0					)
        __UpperCAmelCase			   =     DDPMParallelScheduler()
        torch.manual_seed(0					)
        __UpperCAmelCase			   =     StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     sd_pipe.to(__SCREAMING_SNAKE_CASE					)
        sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE					)

        __UpperCAmelCase			   =     self.get_dummy_inputs(__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     sd_pipe(**__SCREAMING_SNAKE_CASE					).images
        __UpperCAmelCase			   =     image[0, -3:, -3:, -1]
        assert image.shape == (1, 64, 64, 3)

        __UpperCAmelCase			   =     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        lowerCAmelCase_  (self					)       -> List[str]:
        super().test_inference_batch_consistent(batch_sizes=[1, 2]					)


    def        lowerCAmelCase_  (self					)       -> List[str]:
        super().test_inference_batch_single_identical(batch_size=2       , expected_max_diff=3E-3					)






    def        lowerCAmelCase_  (self					)       -> List[str]:
        __UpperCAmelCase			   =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        __UpperCAmelCase			   =     self.get_dummy_components()
        __UpperCAmelCase			   =     StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     sd_pipe.to(__SCREAMING_SNAKE_CASE					)
        sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE					)

        __UpperCAmelCase			   =     self.get_dummy_inputs(__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     '''french fries'''
        __UpperCAmelCase			   =     sd_pipe(**__SCREAMING_SNAKE_CASE       , negative_prompt=__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     output.images
        __UpperCAmelCase			   =     image[0, -3:, -3:, -1]

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

        __UpperCAmelCase			   =     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    A_				( unittest.TestCase  ):


    '''simple docstring'''






    def        lowerCAmelCase_  (self					)       -> Optional[Any]:
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()


    def        lowerCAmelCase_  (self       , lowercase__=0					)       -> str:
        __UpperCAmelCase			   =     torch.Generator(device=__SCREAMING_SNAKE_CASE					).manual_seed(__SCREAMING_SNAKE_CASE					)
        __UpperCAmelCase			   =     {
            '''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        lowerCAmelCase_  (self					)       -> Union[str, Any]:
        __UpperCAmelCase			   =     '''stabilityai/stable-diffusion-2-base'''
        __UpperCAmelCase			   =     DDIMParallelScheduler.from_pretrained(__SCREAMING_SNAKE_CASE       , subfolder='''scheduler'''					)
        __UpperCAmelCase			   =     StableDiffusionParadigmsPipeline.from_pretrained(__SCREAMING_SNAKE_CASE       , scheduler=__SCREAMING_SNAKE_CASE       , safety_checker=__SCREAMING_SNAKE_CASE					)
        pipe.to(__SCREAMING_SNAKE_CASE					)
        pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE					)
        pipe.enable_attention_slicing()

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

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

        __UpperCAmelCase			   =     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







from __future__ import annotations

from collections import deque
from queue import Queue
from timeit import timeit

A_						:     Optional[int]			  =				{
    'A': ['B', 'C'],
    'B': ['A', 'D', 'E'],
    'C': ['A', 'F'],
    'D': ['B'],
    'E': ['B', 'F'],
    'F': ['C', 'E'],
}


def  __a    (				SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE ) ->       list[str]:






    '''simple docstring'''
    __UpperCAmelCase			   =     {start}
    __UpperCAmelCase			   =     [start]
    __UpperCAmelCase			   =     Queue()
    queue.put(SCREAMING_SNAKE_CASE )
    while not queue.empty():
        __UpperCAmelCase			   =     queue.get()
        for w in graph[v]:
            if w not in explored:
                explored.add(SCREAMING_SNAKE_CASE )
                result.append(SCREAMING_SNAKE_CASE )
                queue.put(SCREAMING_SNAKE_CASE )
    return result


def  __a    (				SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE ) ->       list[str]:






    '''simple docstring'''
    __UpperCAmelCase			   =     {start}
    __UpperCAmelCase			   =     [start]
    __UpperCAmelCase			   =     deque([start] )
    while queue:
        __UpperCAmelCase			   =     queue.popleft()
        for child in graph[v]:
            if child not in visited:
                visited.add(SCREAMING_SNAKE_CASE )
                result.append(SCREAMING_SNAKE_CASE )
                queue.append(SCREAMING_SNAKE_CASE )
    return result


def  __a    (				SCREAMING_SNAKE_CASE ) ->       None:






    '''simple docstring'''
    __UpperCAmelCase			   =     f'''from __main__ import G, {name}'''
    __UpperCAmelCase			   =     1_0_0_0_0
    __UpperCAmelCase			   =     timeit(f'''{name}(G, \'A\')'''    ,				setup=SCREAMING_SNAKE_CASE    ,				number=SCREAMING_SNAKE_CASE )
    print(f'''{name:<35} finished {number} runs in {res:.5f} seconds''' )


if __name__ == "__main__":
       import doctest

       doctest.testmod()

       benchmark_function('breadth_first_search')
       benchmark_function('breadth_first_search_with_deque')
       # breadth_first_search                finished 10000 runs in 0.20999 seconds
       # breadth_first_search_with_deque     finished 10000 runs in 0.01421 seconds




