


'''simple docstring'''






import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    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__							(				a__							,							a__							,							unittest.TestCase    ):





							'''simple docstring'''






							lowercase__     :  int      		= StableDiffusionSAGPipeline
							lowercase__     :  Any      		= TEXT_TO_IMAGE_PARAMS
							lowercase__     :  Optional[Any]      		= TEXT_TO_IMAGE_BATCH_PARAMS
							lowercase__     :  Tuple      		= TEXT_TO_IMAGE_IMAGE_PARAMS
							lowercase__     :  Dict      		= TEXT_TO_IMAGE_IMAGE_PARAMS
							lowercase__     :  List[Any]      		= False







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Tuple:
														torch.manual_seed(0  )
														lowerCAmelCase__			  =      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   ,       )
														lowerCAmelCase__			  =      DDIMScheduler(
														    beta_start=0.00_085   ,       beta_end=0.012   ,       beta_schedule='''scaled_linear'''   ,       clip_sample=lowerCamelCase_   ,       set_alpha_to_one=lowerCamelCase_   ,       )
														torch.manual_seed(0  )
														lowerCAmelCase__			  =      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   ,       )
														torch.manual_seed(0  )
														lowerCAmelCase__			  =      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=10_00   ,       )
														lowerCAmelCase__			  =      CLIPTextModel(lowerCamelCase_  )
														lowerCAmelCase__			  =      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''  )

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







							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_=0  )			->				Tuple:
														if str(lowerCamelCase_  ).startswith('''mps'''  ):
																					lowerCAmelCase__			  =      torch.manual_seed(lowerCamelCase_  )
														else:
																					lowerCAmelCase__			  =      torch.Generator(device=lowerCamelCase_  ).manual_seed(lowerCamelCase_  )
														lowerCAmelCase__			  =      {
														    '''prompt''': '''.''',
														    '''generator''': generator,
														    '''num_inference_steps''': 2,
														    '''guidance_scale''': 1.0,
														    '''sag_scale''': 1.0,
														    '''output_type''': '''numpy''',
														}
														return inputs







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




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





							'''simple docstring'''






							def     __SCREAMING_SNAKE_CASE				(						self  )			->				str:
														# clean up the VRAM after each test
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Tuple:
														lowerCAmelCase__			  =      StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''  )
														lowerCAmelCase__			  =      sag_pipe.to(lowerCamelCase_  )
														sag_pipe.set_progress_bar_config(disable=lowerCamelCase_  )

														lowerCAmelCase__			  =      '''.'''
														lowerCAmelCase__			  =      torch.manual_seed(0  )
														lowerCAmelCase__			  =      sag_pipe(
														    [prompt]   ,       generator=lowerCamelCase_   ,       guidance_scale=7.5   ,       sag_scale=1.0   ,       num_inference_steps=20   ,       output_type='''np'''  )

														lowerCAmelCase__			  =      output.images

														lowerCAmelCase__			  =      image[0, -3:, -3:, -1]

														assert image.shape == (1, 5_12, 5_12, 3)
														lowerCAmelCase__			  =      np.array([0.1_568, 0.1_738, 0.1_695, 0.1_693, 0.1_507, 0.1_705, 0.1_547, 0.1_751, 0.1_949]  )

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







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[Any]:
														lowerCAmelCase__			  =      StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''  )
														lowerCAmelCase__			  =      sag_pipe.to(lowerCamelCase_  )
														sag_pipe.set_progress_bar_config(disable=lowerCamelCase_  )

														lowerCAmelCase__			  =      '''.'''
														lowerCAmelCase__			  =      torch.manual_seed(0  )
														lowerCAmelCase__			  =      sag_pipe(
														    [prompt]   ,       generator=lowerCamelCase_   ,       guidance_scale=7.5   ,       sag_scale=1.0   ,       num_inference_steps=20   ,       output_type='''np'''  )

														lowerCAmelCase__			  =      output.images

														lowerCAmelCase__			  =      image[0, -3:, -3:, -1]

														assert image.shape == (1, 5_12, 5_12, 3)
														lowerCAmelCase__			  =      np.array([0.3_459, 0.2_876, 0.2_537, 0.3_002, 0.2_671, 0.2_160, 0.3_026, 0.2_262, 0.2_371]  )

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







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Union[str, Any]:
														lowerCAmelCase__			  =      StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''  )
														lowerCAmelCase__			  =      sag_pipe.to(lowerCamelCase_  )
														sag_pipe.set_progress_bar_config(disable=lowerCamelCase_  )

														lowerCAmelCase__			  =      '''.'''
														lowerCAmelCase__			  =      torch.manual_seed(0  )
														lowerCAmelCase__			  =      sag_pipe(
														    [prompt]   ,       width=7_68   ,       height=5_12   ,       generator=lowerCamelCase_   ,       guidance_scale=7.5   ,       sag_scale=1.0   ,       num_inference_steps=20   ,       output_type='''np'''   ,       )

														lowerCAmelCase__			  =      output.images

														assert image.shape == (1, 5_12, 7_68, 3)







"""simple docstring"""







from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_torch_available, is_torch_tpu_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_torch_available():
				import torch

if is_torch_tpu_available(check_device=False):
				import torch_xla.core.xla_model as xm


_lowerCAmelCase            =	logging.get_logger(__name__)


@dataclass
class      UpperCamelCase		(__snake_case     ):
						_SCREAMING_SNAKE_CASE  : str												=		[
						    """no_inference""",
						    """no_cuda""",
						    """no_tpu""",
						    """no_speed""",
						    """no_memory""",
						    """no_env_print""",
						    """no_multi_process""",
						]






						def __init__(		self  :Union[str, Any]			, **__magic_name__  :int				)      ->Any:
										for deprecated_arg in self.deprecated_args:
														if deprecated_arg in kwargs:
																		lowercase    :      int       = deprecated_arg[3:]
																		setattr(self			, __magic_name__			, not kwargs.pop(__magic_name__				)				)
																		logger.warning(
																		    f"""{deprecated_arg} is depreciated. Please use --no_{positive_arg} or"""
																		    f""" {positive_arg}={kwargs[positive_arg]}"""				)

										lowercase    :      List[str]       = kwargs.pop("""torchscript"""			, self.torchscript				)
										lowercase    :      Any       = kwargs.pop("""torch_xla_tpu_print_metrics"""			, self.torch_xla_tpu_print_metrics				)
										lowercase    :      Union[str, Any]       = kwargs.pop("""fp16_opt_level"""			, self.fpaa_opt_level				)
										super().__init__(**__magic_name__				)

						_SCREAMING_SNAKE_CASE  : bool												=		field(default=__snake_case						,						metadata={"""help""": """Trace the models using torchscript"""}     )
						_SCREAMING_SNAKE_CASE  : bool												=		field(default=__snake_case						,						metadata={"""help""": """Print Xla/PyTorch tpu metrics"""}     )
						_SCREAMING_SNAKE_CASE  : str												=		field(
						    default="""O1"""						,						metadata={
						        """help""": (
						            """For fp16: Apex AMP optimization level selected in ['O0', 'O1', 'O2', and 'O3']. """
						            """See details at https://nvidia.github.io/apex/amp.html"""
						        )
						    }						,						)






						@cached_property
						def      __snake_case      (		self  :Optional[int]				)      ->Tuple["torch.device", int]:
										requires_backends(self			, ["""torch"""]				)
										logger.info("""PyTorch: setting up devices"""				)
										if not self.cuda:
														lowercase    :      Optional[Any]       = torch.device("""cpu"""				)
														lowercase    :      int       = 0
										elif is_torch_tpu_available():
														lowercase    :      Any       = xm.xla_device()
														lowercase    :      List[str]       = 0
										else:
														lowercase    :      int       = torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""				)
														lowercase    :      Any       = torch.cuda.device_count()
										return device, n_gpu






						@property
						def      __snake_case      (		self  :Optional[int]				)      ->Union[str, Any]:
										return is_torch_tpu_available() and self.tpu






						@property
						def      __snake_case      (		self  :Optional[int]				)      ->int:
										requires_backends(self			, ["""torch"""]				)
										# TODO(PVP): currently only single GPU is supported
										return torch.cuda.current_device()






						@property
						def      __snake_case      (		self  :Optional[int]				)      ->"torch.device":
										requires_backends(self			, ["""torch"""]				)
										return self._setup_devices[0]






						@property
						def      __snake_case      (		self  :Optional[Any]				)      ->str:
										requires_backends(self			, ["""torch"""]				)
										return self._setup_devices[1]






						@property
						def      __snake_case      (		self  :int				)      ->List[str]:
										return self.n_gpu > 0
