


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				(							_lowercase	,		_lowercase	,		unittest.TestCase):
	_a    :    str        =				StableDiffusionParadigmsPipeline
	_a    :    Optional[Any]        =				TEXT_TO_IMAGE_PARAMS
	_a    :    Any        =				TEXT_TO_IMAGE_BATCH_PARAMS
	_a    :    List[Any]        =				TEXT_TO_IMAGE_IMAGE_PARAMS
	_a    :    Optional[Any]        =				TEXT_TO_IMAGE_IMAGE_PARAMS




	def 				UpperCAmelCase__(   self		:						List[Any]						)->			Any:
			torch.manual_seed(0						)
			lowerCAmelCase__  :     Optional[int]       	=   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__  :     str       	=   DDIMParallelScheduler(
			    beta_start=0.0_0085						,   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__  :     Tuple       	=   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__  :     Optional[int]       	=   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__  :     int       	=   CLIPTextModel(_SCREAMING_SNAKE_CASE						)
			lowerCAmelCase__  :     List[str]       	=   CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

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




	def 				UpperCAmelCase__(   self		:						int						,   _SCREAMING_SNAKE_CASE		:						Optional[int]						,   _SCREAMING_SNAKE_CASE		:						Any=0						)->			str:
			if str(_SCREAMING_SNAKE_CASE						).startswith('''mps'''						):
					lowerCAmelCase__  :     List[str]       	=   torch.manual_seed(_SCREAMING_SNAKE_CASE						)
			else:
					lowerCAmelCase__  :     List[Any]       	=   torch.Generator(device=_SCREAMING_SNAKE_CASE						).manual_seed(_SCREAMING_SNAKE_CASE						)
			lowerCAmelCase__  :     Tuple       	=   {
			    '''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 				UpperCAmelCase__(   self		:						Dict						)->			Optional[int]:
			lowerCAmelCase__  :     Dict       	=   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
			lowerCAmelCase__  :     Dict       	=   self.get_dummy_components()
			lowerCAmelCase__  :     List[str]       	=   StableDiffusionParadigmsPipeline(**_SCREAMING_SNAKE_CASE						)
			lowerCAmelCase__  :     Tuple       	=   sd_pipe.to(_SCREAMING_SNAKE_CASE						)
			sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE						)

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

			lowerCAmelCase__  :     Tuple       	=   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 				UpperCAmelCase__(   self		:						Optional[Any]						)->			List[Any]:
			lowerCAmelCase__  :     Any       	=   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
			lowerCAmelCase__  :     List[str]       	=   self.get_dummy_components()
			torch.manual_seed(0						)
			lowerCAmelCase__  :     int       	=   DDPMParallelScheduler()
			torch.manual_seed(0						)
			lowerCAmelCase__  :     List[str]       	=   StableDiffusionParadigmsPipeline(**_SCREAMING_SNAKE_CASE						)
			lowerCAmelCase__  :     List[str]       	=   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__  :     List[str]       	=   sd_pipe(**_SCREAMING_SNAKE_CASE						).images
			lowerCAmelCase__  :     Any       	=   image[0, -3:, -3:, -1]
			assert image.shape == (1, 64, 64, 3)

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




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




	def 				UpperCAmelCase__(   self		:						Optional[int]						)->			List[Any]:
			lowerCAmelCase__  :     List[Any]       	=   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
			lowerCAmelCase__  :     Tuple       	=   self.get_dummy_components()
			lowerCAmelCase__  :     Optional[Any]       	=   StableDiffusionParadigmsPipeline(**_SCREAMING_SNAKE_CASE						)
			lowerCAmelCase__  :     Optional[Any]       	=   sd_pipe.to(_SCREAMING_SNAKE_CASE						)
			sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE						)

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

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

			lowerCAmelCase__  :     Union[str, Any]       	=   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):




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




	def 				UpperCAmelCase__(   self		:						Dict						,   _SCREAMING_SNAKE_CASE		:						int=0						)->			Dict:
			lowerCAmelCase__  :     int       	=   torch.Generator(device=_SCREAMING_SNAKE_CASE						).manual_seed(_SCREAMING_SNAKE_CASE						)
			lowerCAmelCase__  :     Optional[int]       	=   {
			    '''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 				UpperCAmelCase__(   self		:						Optional[Any]						)->			Dict:
			lowerCAmelCase__  :     Optional[int]       	=   '''stabilityai/stable-diffusion-2-base'''
			lowerCAmelCase__  :     List[Any]       	=   DDIMParallelScheduler.from_pretrained(_SCREAMING_SNAKE_CASE						,   subfolder='''scheduler'''						)
			lowerCAmelCase__  :     int       	=   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__  :     Optional[Any]       	=   self.get_inputs()
			lowerCAmelCase__  :     List[Any]       	=   pipe(**_SCREAMING_SNAKE_CASE						).images
			lowerCAmelCase__  :     List[str]       	=   image[0, -3:, -3:, -1].flatten()

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

			lowerCAmelCase__  :     Dict       	=   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








import os

from packaging import version

from .. import __version__
from .accelerate_utils import apply_forward_hook
from .constants import (
    CONFIG_NAME,
    DEPRECATED_REVISION_ARGS,
    DIFFUSERS_CACHE,
    DIFFUSERS_DYNAMIC_MODULE_NAME,
    FLAX_WEIGHTS_NAME,
    HF_MODULES_CACHE,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    ONNX_EXTERNAL_WEIGHTS_NAME,
    ONNX_WEIGHTS_NAME,
    SAFETENSORS_WEIGHTS_NAME,
    TEXT_ENCODER_ATTN_MODULE,
    WEIGHTS_NAME,
)
from .deprecation_utils import deprecate
from .doc_utils import replace_example_docstring
from .dynamic_modules_utils import get_class_from_dynamic_module
from .hub_utils import (
    HF_HUB_OFFLINE,
    _add_variant,
    _get_model_file,
    extract_commit_hash,
    http_user_agent,
)
from .import_utils import (
    BACKENDS_MAPPING,
    ENV_VARS_TRUE_AND_AUTO_VALUES,
    ENV_VARS_TRUE_VALUES,
    USE_JAX,
    USE_TF,
    USE_TORCH,
    DummyObject,
    OptionalDependencyNotAvailable,
    is_accelerate_available,
    is_accelerate_version,
    is_bsa_available,
    is_flax_available,
    is_ftfy_available,
    is_inflect_available,
    is_invisible_watermark_available,
    is_k_diffusion_available,
    is_k_diffusion_version,
    is_librosa_available,
    is_note_seq_available,
    is_omegaconf_available,
    is_onnx_available,
    is_safetensors_available,
    is_scipy_available,
    is_tensorboard_available,
    is_tf_available,
    is_torch_available,
    is_torch_version,
    is_torchsde_available,
    is_transformers_available,
    is_transformers_version,
    is_unidecode_available,
    is_wandb_available,
    is_xformers_available,
    requires_backends,
)
from .logging import get_logger
from .outputs import BaseOutput
from .pil_utils import PIL_INTERPOLATION, numpy_to_pil, pt_to_pil
from .torch_utils import is_compiled_module, randn_tensor


if is_torch_available():
		from .testing_utils import (
		    floats_tensor,
		    load_hf_numpy,
		    load_image,
		    load_numpy,
		    load_pt,
		    nightly,
		    parse_flag_from_env,
		    print_tensor_test,
		    require_torch_a,
		    require_torch_gpu,
		    skip_mps,
		    slow,
		    torch_all_close,
		    torch_device,
		)
		from .torch_utils import maybe_allow_in_graph

from .testing_utils import export_to_gif, export_to_video


lowerCamelCase		=						get_logger(__name__)






def 	lowerCamelCase_       (			_a			):




		"""simple docstring"""






		if version.parse(_a			) < version.parse(_a			):
				if "dev" in min_version:
						lowerCAmelCase__  :     Tuple       	=   (
						    '''This example requires a source install from HuggingFace diffusers (see '''
						    '''`https://huggingface.co/docs/diffusers/installation#install-from-source`),'''
						)
				else:
						lowerCAmelCase__  :     Dict       	=   f'This example requires a minimum version of {min_version},'
				error_message += f' but the version found is {__version__}.\n'
				raise ImportError(_a			)




