


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 						lowerCamelCase_							(					lowerCAmelCase__  ,  lowerCAmelCase__  ,  unittest.TestCase					):



					'''simple docstring'''







					__UpperCAmelCase							  =					StableDiffusionSAGPipeline
					__UpperCAmelCase							  =					TEXT_TO_IMAGE_PARAMS
					__UpperCAmelCase							  =					TEXT_TO_IMAGE_BATCH_PARAMS
					__UpperCAmelCase							  =					TEXT_TO_IMAGE_IMAGE_PARAMS
					__UpperCAmelCase							  =					TEXT_TO_IMAGE_IMAGE_PARAMS
					__UpperCAmelCase							  =					False






					def A   (					self    ) ->      Optional[int]:







								'''simple docstring'''

								torch.manual_seed(0    )
								__lowercase			       = UNetaDConditionModel(
								    block_out_channels=(3_2, 6_4)			,							layers_per_block=2			,							sample_size=3_2			,							in_channels=4			,							out_channels=4			,							down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')			,							up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')			,							cross_attention_dim=3_2			,							)
								__lowercase			       = DDIMScheduler(
								    beta_start=0.0_0_0_8_5			,							beta_end=0.0_1_2			,							beta_schedule='''scaled_linear'''			,							clip_sample=snake_case_			,							set_alpha_to_one=snake_case_			,							)
								torch.manual_seed(0    )
								__lowercase			       = AutoencoderKL(
								    block_out_channels=[3_2, 6_4]			,							in_channels=3			,							out_channels=3			,							down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']			,							up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']			,							latent_channels=4			,							)
								torch.manual_seed(0    )
								__lowercase			       = CLIPTextConfig(
								    bos_token_id=0			,							eos_token_id=2			,							hidden_size=3_2			,							intermediate_size=3_7			,							layer_norm_eps=1e-0_5			,							num_attention_heads=4			,							num_hidden_layers=5			,							pad_token_id=1			,							vocab_size=1_0_0_0			,							)
								__lowercase			       = CLIPTextModel(snake_case_    )
								__lowercase			       = CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''    )

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






					def A   (					self			,							snake_case_			,							snake_case_=0    ) ->      Optional[Any]:







								'''simple docstring'''

								if str(snake_case_    ).startswith('''mps'''    ):
											__lowercase			       = torch.manual_seed(snake_case_    )
								else:
											__lowercase			       = torch.Generator(device=snake_case_    ).manual_seed(snake_case_    )
								__lowercase			       = {
								    '''prompt''': '''.''',
								    '''generator''': generator,
								    '''num_inference_steps''': 2,
								    '''guidance_scale''': 1.0,
								    '''sag_scale''': 1.0,
								    '''output_type''': '''numpy''',
								}
								return inputs






					def A   (					self    ) ->      Any:







								'''simple docstring'''

								super().test_inference_batch_single_identical(expected_max_diff=3e-3    )




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



					'''simple docstring'''







					def A   (					self    ) ->      str:







								'''simple docstring'''

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






					def A   (					self    ) ->      List[Any]:







								'''simple docstring'''

								__lowercase			       = StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''    )
								__lowercase			       = sag_pipe.to(snake_case_    )
								sag_pipe.set_progress_bar_config(disable=snake_case_    )

								__lowercase			       = '''.'''
								__lowercase			       = torch.manual_seed(0    )
								__lowercase			       = sag_pipe(
								    [prompt]			,							generator=snake_case_			,							guidance_scale=7.5			,							sag_scale=1.0			,							num_inference_steps=2_0			,							output_type='''np'''    )

								__lowercase			       = output.images

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

								assert image.shape == (1, 5_1_2, 5_1_2, 3)
								__lowercase			       = np.array([0.1_5_6_8, 0.1_7_3_8, 0.1_6_9_5, 0.1_6_9_3, 0.1_5_0_7, 0.1_7_0_5, 0.1_5_4_7, 0.1_7_5_1, 0.1_9_4_9]    )

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






					def A   (					self    ) ->      Dict:







								'''simple docstring'''

								__lowercase			       = StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''    )
								__lowercase			       = sag_pipe.to(snake_case_    )
								sag_pipe.set_progress_bar_config(disable=snake_case_    )

								__lowercase			       = '''.'''
								__lowercase			       = torch.manual_seed(0    )
								__lowercase			       = sag_pipe(
								    [prompt]			,							generator=snake_case_			,							guidance_scale=7.5			,							sag_scale=1.0			,							num_inference_steps=2_0			,							output_type='''np'''    )

								__lowercase			       = output.images

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

								assert image.shape == (1, 5_1_2, 5_1_2, 3)
								__lowercase			       = np.array([0.3_4_5_9, 0.2_8_7_6, 0.2_5_3_7, 0.3_0_0_2, 0.2_6_7_1, 0.2_1_6_0, 0.3_0_2_6, 0.2_2_6_2, 0.2_3_7_1]    )

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






					def A   (					self    ) ->      List[str]:







								'''simple docstring'''

								__lowercase			       = StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''    )
								__lowercase			       = sag_pipe.to(snake_case_    )
								sag_pipe.set_progress_bar_config(disable=snake_case_    )

								__lowercase			       = '''.'''
								__lowercase			       = torch.manual_seed(0    )
								__lowercase			       = sag_pipe(
								    [prompt]			,							width=7_6_8			,							height=5_1_2			,							generator=snake_case_			,							guidance_scale=7.5			,							sag_scale=1.0			,							num_inference_steps=2_0			,							output_type='''np'''			,							)

								__lowercase			       = output.images

								assert image.shape == (1, 5_1_2, 7_6_8, 3)






from typing import Optional

from .. import Features, NamedSplit
from ..packaged_modules.text.text import Text
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader
class 						lowerCamelCase_							(					lowerCAmelCase__					):



					'''simple docstring'''







					def __init__(					self			,							snake_case_			,							snake_case_ = None			,							snake_case_ = None			,							snake_case_ = None			,							snake_case_ = False			,							snake_case_ = False			,							snake_case_ = None			,							**snake_case_			,							) ->      Dict:







								'''simple docstring'''

								super().__init__(
								    snake_case_			,							split=snake_case_			,							features=snake_case_			,							cache_dir=snake_case_			,							keep_in_memory=snake_case_			,							streaming=snake_case_			,							num_proc=snake_case_			,							**snake_case_			,							)
								__lowercase			       = path_or_paths if isinstance(snake_case_			,							snake_case_    ) else {self.split: path_or_paths}
								__lowercase			       = Text(
								    cache_dir=snake_case_			,							data_files=snake_case_			,							features=snake_case_			,							**snake_case_			,							)






					def A   (					self    ) ->      int:







								'''simple docstring'''

								if self.streaming:
											__lowercase			       = self.builder.as_streaming_dataset(split=self.split    )
								# Build regular (map-style) dataset
								else:
											__lowercase			       = None
											__lowercase			       = None
											__lowercase			       = None
											__lowercase			       = None

											self.builder.download_and_prepare(
											    download_config=snake_case_			,							download_mode=snake_case_			,							verification_mode=snake_case_			,							base_path=snake_case_			,							num_proc=self.num_proc			,							)
											__lowercase			       = self.builder.as_dataset(
											    split=self.split			,							verification_mode=snake_case_			,							in_memory=self.keep_in_memory    )
								return dataset


