


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

							"""simple docstring"""


							SCREAMING_SNAKE_CASE   :					List[Any]        =						StableDiffusionParadigmsPipeline
							SCREAMING_SNAKE_CASE   :					Optional[Any]        =						TEXT_TO_IMAGE_PARAMS
							SCREAMING_SNAKE_CASE   :					Dict        =						TEXT_TO_IMAGE_BATCH_PARAMS
							SCREAMING_SNAKE_CASE   :					str        =						TEXT_TO_IMAGE_IMAGE_PARAMS
							SCREAMING_SNAKE_CASE   :					List[Any]        =						TEXT_TO_IMAGE_IMAGE_PARAMS





							def 		__magic_name__			(    self	)->	Optional[int]:
											torch.manual_seed(0	)
											_SCREAMING_SNAKE_CASE      = 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=A_				, )
											_SCREAMING_SNAKE_CASE      = DDIMParallelScheduler(
											    beta_start=0.0_0085				, beta_end=0.012				, beta_schedule='scaled_linear'				, clip_sample=A_				, set_alpha_to_one=A_				, )
											torch.manual_seed(0	)
											_SCREAMING_SNAKE_CASE      = 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	)
											_SCREAMING_SNAKE_CASE      = CLIPTextConfig(
											    bos_token_id=0				, eos_token_id=2				, hidden_size=32				, intermediate_size=37				, layer_norm_eps=1e-0_5				, num_attention_heads=4				, num_hidden_layers=5				, pad_token_id=1				, vocab_size=1000				, hidden_act='gelu'				, projection_dim=512				, )
											_SCREAMING_SNAKE_CASE      = CLIPTextModel(A_	)
											_SCREAMING_SNAKE_CASE      = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'	)

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





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

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

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

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

											_SCREAMING_SNAKE_CASE      = 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 		__magic_name__			(    self	)->	Any:
											super().test_inference_batch_consistent(batch_sizes=[1, 2]	)





							def 		__magic_name__			(    self	)->	int:
											super().test_inference_batch_single_identical(batch_size=2				, expected_max_diff=3e-3	)





							def 		__magic_name__			(    self	)->	Tuple:
											_SCREAMING_SNAKE_CASE      = 'cpu'  # ensure determinism for the device-dependent torch.Generator
											_SCREAMING_SNAKE_CASE      = self.get_dummy_components()
											_SCREAMING_SNAKE_CASE      = StableDiffusionParadigmsPipeline(**A_	)
											_SCREAMING_SNAKE_CASE      = sd_pipe.to(A_	)
											sd_pipe.set_progress_bar_config(disable=A_	)

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

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

											_SCREAMING_SNAKE_CASE      = 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    __lowercase  (       unittest.TestCase							):

							"""simple docstring"""


							def 		__magic_name__			(    self	)->	Tuple:
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()





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

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

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

											_SCREAMING_SNAKE_CASE      = 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 sys
from dataclasses import dataclass
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union

import pandas as pd
import pyarrow as pa

import datasets
import datasets.config
from datasets.features.features import require_storage_cast
from datasets.table import table_cast


if TYPE_CHECKING:
							import sqlitea

							import sqlalchemy


snake_case			:      List[str] 							=   datasets.utils.logging.get_logger(__name__)



@dataclass
class    __lowercase  (       datasets.BuilderConfig							):

							"""simple docstring"""


							SCREAMING_SNAKE_CASE   :					Union[str, "sqlalchemy.sql.Selectable"]        =						None
							SCREAMING_SNAKE_CASE   :					Union[str, "sqlalchemy.engine.Connection", "sqlalchemy.engine.Engine", "sqlite3.Connection"]        =						None
							SCREAMING_SNAKE_CASE   :					Optional[Union[str, List[str]]]        =						None
							SCREAMING_SNAKE_CASE   :					bool        =						True
							SCREAMING_SNAKE_CASE   :					Optional[Union[List, Tuple, Dict]]        =						None
							SCREAMING_SNAKE_CASE   :					Optional[Union[List, Dict]]        =						None
							SCREAMING_SNAKE_CASE   :					Optional[List[str]]        =						None
							SCREAMING_SNAKE_CASE   :					Optional[int]        =						1_00_00
							SCREAMING_SNAKE_CASE   :					Optional[datasets.Features]        =						None





							def 		__magic_name__			(    self	)->	int:
											if self.sql is None:
															raise ValueError('sql must be specified'	)
											if self.con is None:
															raise ValueError('con must be specified'	)





							def 		__magic_name__			(    self				, A_				, A_ = None				, )->	str:
											_SCREAMING_SNAKE_CASE      = config_kwargs.copy()
											# We need to stringify the Selectable object to make its hash deterministic

											# The process of stringifying is explained here: http://docs.sqlalchemy.org/en/latest/faq/sqlexpressions.html
											_SCREAMING_SNAKE_CASE      = config_kwargs['sql']
											if not isinstance(A_				, A_	):
															if datasets.config.SQLALCHEMY_AVAILABLE and "sqlalchemy" in sys.modules:
																			import sqlalchemy

																			if isinstance(A_				, sqlalchemy.sql.Selectable	):
																							_SCREAMING_SNAKE_CASE      = sqlalchemy.create_engine(config_kwargs['con'].split('://'	)[0] + '://'	)
																							_SCREAMING_SNAKE_CASE      = str(sql.compile(dialect=engine.dialect	)	)
																							_SCREAMING_SNAKE_CASE      = sql_str
																			else:
																							raise TypeError(
																							    F'''Supported types for \'sql\' are string and sqlalchemy.sql.Selectable but got {type(A_	)}: {sql}'''	)
															else:
																			raise TypeError(
																			    F'''Supported types for \'sql\' are string and sqlalchemy.sql.Selectable but got {type(A_	)}: {sql}'''	)
											_SCREAMING_SNAKE_CASE      = config_kwargs['con']
											if not isinstance(A_				, A_	):
															_SCREAMING_SNAKE_CASE      = id(A_	)
															logger.info(
															    F'''SQL connection \'con\' of type {type(A_	)} couldn\'t be hashed properly. To enable hashing, specify \'con\' as URI string instead.'''	)

											return super().create_config_id(A_				, custom_features=A_	)





							@property
							def 		__magic_name__			(    self	)->	Dict:
											_SCREAMING_SNAKE_CASE      = {
											    'index_col': self.index_col,
											    'columns': self.columns,
											    'params': self.params,
											    'coerce_float': self.coerce_float,
											    'parse_dates': self.parse_dates,
											}
											return pd_read_sql_kwargs







class    __lowercase  (       datasets.ArrowBasedBuilder							):

							"""simple docstring"""


							SCREAMING_SNAKE_CASE   :					Any        =						SqlConfig





							def 		__magic_name__			(    self	)->	Dict:
											return datasets.DatasetInfo(features=self.config.features	)





							def 		__magic_name__			(    self				, A_	)->	Optional[Any]:
											return [datasets.SplitGenerator(name=datasets.Split.TRAIN				, gen_kwargs={}	)]





							def 		__magic_name__			(    self				, A_	)->	pa.Table:
											if self.config.features is not None:
															_SCREAMING_SNAKE_CASE      = self.config.features.arrow_schema
															if all(not require_storage_cast(A_	) for feature in self.config.features.values()	):
																			# cheaper cast
																			_SCREAMING_SNAKE_CASE      = pa.Table.from_arrays([pa_table[field.name] for field in schema]				, schema=A_	)
															else:
																			# more expensive cast; allows str <-> int/float or str to Audio for example
																			_SCREAMING_SNAKE_CASE      = table_cast(A_				, A_	)
											return pa_table





							def 		__magic_name__			(    self	)->	Optional[int]:
											_SCREAMING_SNAKE_CASE      = self.config.chunksize
											_SCREAMING_SNAKE_CASE      = pd.read_sql(
											    self.config.sql				, self.config.con				, chunksize=A_				, **self.config.pd_read_sql_kwargs	)
											_SCREAMING_SNAKE_CASE      = [sql_reader] if chunksize is None else sql_reader
											for chunk_idx, df in enumerate(A_	):
															_SCREAMING_SNAKE_CASE      = pa.Table.from_pandas(A_	)
															yield chunk_idx, self._cast_table(A_	)
