
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 						lowercase       (  A_	,		A_	,		unittest.TestCase					):
     __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 lowercase_			(		self    ):
        """simple docstring"""







        torch.manual_seed(0    )
        lowerCAmelCase__ :    Dict				      =			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__ ,	)
        lowerCAmelCase__ :    Optional[Any]				      =			DDIMParallelScheduler(
            beta_start=0.00_085 ,	beta_end=0.012 ,	beta_schedule='''scaled_linear''' ,	clip_sample=SCREAMING_SNAKE_CASE__ ,	set_alpha_to_one=SCREAMING_SNAKE_CASE__ ,	)
        torch.manual_seed(0    )
        lowerCAmelCase__ :    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    )
        lowerCAmelCase__ :    Tuple				      =			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=512 ,	)
        lowerCAmelCase__ :    Union[str, Any]				      =			CLIPTextModel(SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    List[str]				      =			CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''    )

        lowerCAmelCase__ :    Optional[int]				      =			{
            '''unet''': unet,
            '''scheduler''': scheduler,
            '''vae''': vae,
            '''text_encoder''': text_encoder,
            '''tokenizer''': tokenizer,
            '''safety_checker''': None,
            '''feature_extractor''': None,
        }
        return components
     def lowercase_			(		self ,	SCREAMING_SNAKE_CASE__ ,	SCREAMING_SNAKE_CASE__=0    ):
        """simple docstring"""







        if str(SCREAMING_SNAKE_CASE__    ).startswith('''mps'''    ):
           lowerCAmelCase__ :    int				      =			torch.manual_seed(SCREAMING_SNAKE_CASE__    )
        else:
           lowerCAmelCase__ :    Union[str, Any]				      =			torch.Generator(device=SCREAMING_SNAKE_CASE__    ).manual_seed(SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    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 lowercase_			(		self    ):
        """simple docstring"""







        lowerCAmelCase__ :    str				      =			'''cpu'''  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__ :    List[str]				      =			self.get_dummy_components()
        lowerCAmelCase__ :    str				      =			StableDiffusionParadigmsPipeline(**SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    int				      =			sd_pipe.to(SCREAMING_SNAKE_CASE__    )
        sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__    )

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

        lowerCAmelCase__ :    Tuple				      =			np.array([0.4_773, 0.5_417, 0.4_723, 0.4_925, 0.5_631, 0.4_752, 0.5_240, 0.4_935, 0.5_023]    )

        assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1E-2
     def lowercase_			(		self    ):
        """simple docstring"""







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

        lowerCAmelCase__ :    Union[str, Any]				      =			self.get_dummy_inputs(SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    Optional[int]				      =			sd_pipe(**SCREAMING_SNAKE_CASE__    ).images
        lowerCAmelCase__ :    Tuple				      =			image[0, -3:, -3:, -1]
        assert image.shape == (1, 64, 64, 3)

        lowerCAmelCase__ :    Optional[Any]				      =			np.array([0.3_573, 0.4_420, 0.4_960, 0.4_799, 0.3_796, 0.3_879, 0.4_819, 0.4_365, 0.4_468]    )

        assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1E-2
     def lowercase_			(		self    ):
        """simple docstring"""







        super().test_inference_batch_consistent(batch_sizes=[1, 2]    )
     def lowercase_			(		self    ):
        """simple docstring"""







        super().test_inference_batch_single_identical(batch_size=2 ,	expected_max_diff=3E-3    )
     def lowercase_			(		self    ):
        """simple docstring"""







        lowerCAmelCase__ :    str				      =			'''cpu'''  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__ :    Tuple				      =			self.get_dummy_components()
        lowerCAmelCase__ :    Tuple				      =			StableDiffusionParadigmsPipeline(**SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    List[Any]				      =			sd_pipe.to(SCREAMING_SNAKE_CASE__    )
        sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__    )

        lowerCAmelCase__ :    Any				      =			self.get_dummy_inputs(SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    Dict				      =			'''french fries'''
        lowerCAmelCase__ :    List[Any]				      =			sd_pipe(**SCREAMING_SNAKE_CASE__ ,	negative_prompt=SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    Any				      =			output.images
        lowerCAmelCase__ :    Tuple				      =			image[0, -3:, -3:, -1]

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

        lowerCAmelCase__ :    List[Any]				      =			np.array([0.4_771, 0.5_420, 0.4_683, 0.4_918, 0.5_636, 0.4_725, 0.5_230, 0.4_923, 0.5_015]    )

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





@slow
@require_torch_gpu
class 						lowercase       (  unittest.TestCase					):
     def lowercase_			(		self    ):
        """simple docstring"""







        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()
     def lowercase_			(		self ,	SCREAMING_SNAKE_CASE__=0    ):
        """simple docstring"""







        lowerCAmelCase__ :    int				      =			torch.Generator(device=SCREAMING_SNAKE_CASE__    ).manual_seed(SCREAMING_SNAKE_CASE__    )
        lowerCAmelCase__ :    List[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 lowercase_			(		self    ):
        """simple docstring"""







        lowerCAmelCase__ :    Dict				      =			'''stabilityai/stable-diffusion-2-base'''
        lowerCAmelCase__ :    str				      =			DDIMParallelScheduler.from_pretrained(SCREAMING_SNAKE_CASE__ ,	subfolder='''scheduler'''    )
        lowerCAmelCase__ :    List[str]				      =			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()

        lowerCAmelCase__ :    Tuple				      =			self.get_inputs()
        lowerCAmelCase__ :    Dict				      =			pipe(**SCREAMING_SNAKE_CASE__    ).images
        lowerCAmelCase__ :    Tuple				      =			image[0, -3:, -3:, -1].flatten()

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

        lowerCAmelCase__ :    Tuple				      =			np.array([0.9_622, 0.9_602, 0.9_748, 0.9_591, 0.9_630, 0.9_691, 0.9_661, 0.9_631, 0.9_741]    )

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





import abc
import copy
import dataclasses
from dataclasses import dataclass
from typing import ClassVar, Dict, Type, TypeVar

from ..features import Features


A__       :   Tuple   						=					TypeVar("""T""", bound="""TaskTemplate""")





@dataclass(frozen=__UpperCamelCase					)
class 						lowercase       (  abc.ABC					):
					# `task` is not a ClassVar since we want it to be part of the `asdict` output for JSON serialization
					__a					=    42
					__a					=    42
					__a					=    42
					def lowercase_			(		self ,	SCREAMING_SNAKE_CASE__    ):
								"""simple docstring"""







								return copy.deepcopy(self    )
					@property
					def lowercase_			(		self    ):
								"""simple docstring"""







								return Features(**self.input_schema ,	**self.label_schema    )
					@property
					@abc.abstractmethod
					def lowercase_			(		self    ):
								"""simple docstring"""







								raise NotImplementedError
					@classmethod
					def lowercase_			(		cls ,	SCREAMING_SNAKE_CASE__    ):
								"""simple docstring"""







								lowerCAmelCase__ :    List[Any]				      =			{f.name for f in dataclasses.fields(cls    )}
								return cls(**{k: v for k, v in template_dict.items() if k in field_names}    )



