



'''simple docstring'''

import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import 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 PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


__lowerCAmelCase									=   False

@skip_mps
class 	__magic_name__						(    _UpperCamelCase			,						_UpperCamelCase			,						_UpperCamelCase			,						unittest.TestCase   ):
							lowerCAmelCase       :    List[Any]   				=					StableDiffusionAttendAndExcitePipeline
							lowerCAmelCase       :    str   				=					False
							lowerCAmelCase       :    Optional[Any]   				=					TEXT_TO_IMAGE_PARAMS
							lowerCAmelCase       :    Optional[Any]   				=					TEXT_TO_IMAGE_BATCH_PARAMS.union({'token_indices'}   )
							lowerCAmelCase       :    str   				=					TEXT_TO_IMAGE_IMAGE_PARAMS
							lowerCAmelCase       :    List[Any]   				=					TEXT_TO_IMAGE_IMAGE_PARAMS







							@classmethod
							def 							__lowercase			(							cls					:       Tuple		):
									super().setUpClass()
									torch.use_deterministic_algorithms(_UpperCAmelCase		)







							@classmethod
							def 							__lowercase			(							cls					:       List[str]		):
									super().tearDownClass()
									torch.use_deterministic_algorithms(_UpperCAmelCase		)







							def 							__lowercase			(							self					:       str		):
									torch.manual_seed(0		)
									_a     :      int        =							UNetaDConditionModel(
									    block_out_channels=(32, 64)	,layers_per_block=1	,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=_UpperCAmelCase	,)
									_a     :      Dict        =							DDIMScheduler(
									    beta_start=0.0_00_85	,beta_end=0.0_12	,beta_schedule='scaled_linear'	,clip_sample=_UpperCAmelCase	,set_alpha_to_one=_UpperCAmelCase	,)
									torch.manual_seed(0		)
									_a     :      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		)
									_a     :      Any        =							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	,)
									_a     :      List[Any]        =							CLIPTextModel(_UpperCAmelCase		)
									_a     :      List[Any]        =							CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

									_a     :      int        =							{
									    'unet': unet,
									    'scheduler': scheduler,
									    'vae': vae,
									    'text_encoder': text_encoder,
									    'tokenizer': tokenizer,
									    'safety_checker': None,
									    'feature_extractor': None,
									}

									return components







							def 							__lowercase			(							self					:       Optional[Any]	,_UpperCAmelCase					:       Optional[int]	,_UpperCAmelCase					:       Optional[int]=0		):
									if str(_UpperCAmelCase		).startswith('mps'		):
											_a     :      Union[str, Any]        =							torch.manual_seed(_UpperCAmelCase		)
									else:
											_a     :      Tuple        =							torch.Generator(device=_UpperCAmelCase		).manual_seed(_UpperCAmelCase		)
									_a     :      Union[str, Any]        =							{
									    'prompt': 'a cat and a frog',
									    'token_indices': [2, 5],
									    'generator': generator,
									    'num_inference_steps': 1,
									    'guidance_scale': 6.0,
									    'output_type': 'numpy',
									    'max_iter_to_alter': 2,
									    'thresholds': {0: 0.7},
									}
									return inputs







							def 							__lowercase			(							self					:       Union[str, Any]		):
									_a     :      Any        =							'cpu'

									_a     :      Optional[Any]        =							self.get_dummy_components()
									_a     :      Optional[Any]        =							self.pipeline_class(**_UpperCAmelCase		)
									pipe.to(_UpperCAmelCase		)
									pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

									_a     :      Dict        =							self.get_dummy_inputs(_UpperCAmelCase		)
									_a     :      int        =							pipe(**_UpperCAmelCase		).images
									_a     :      Dict        =							image[0, -3:, -3:, -1]

									self.assertEqual(image.shape	,(1, 64, 64, 3)		)
									_a     :      int        =							np.array(
									    [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96]		)
									_a     :      Dict        =							np.abs(image_slice.flatten() - expected_slice		).max()
									self.assertLessEqual(_UpperCAmelCase	,1E-3		)







							def 							__lowercase			(							self					:       Union[str, Any]		):
									super().test_cpu_offload_forward_pass(expected_max_diff=5E-4		)







							def 							__lowercase			(							self					:       List[Any]		):
									# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
									self._test_inference_batch_consistent(batch_sizes=[1, 2]		)







							def 							__lowercase			(							self					:       Union[str, Any]		):
									self._test_inference_batch_single_identical(batch_size=2	,expected_max_diff=7E-4		)







							def 							__lowercase			(							self					:       Any		):
									super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3		)







							def 							__lowercase			(							self					:       Optional[Any]		):
									super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5E-4		)







							def 							__lowercase			(							self					:       Optional[Any]		):
									super().test_save_load_local(expected_max_difference=5E-4		)







							def 							__lowercase			(							self					:       List[str]		):
									super().test_save_load_optional_components(expected_max_difference=4E-4		)

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







							@classmethod
							def 							__lowercase			(							cls					:       str		):
									super().setUpClass()
									torch.use_deterministic_algorithms(_UpperCAmelCase		)







							@classmethod
							def 							__lowercase			(							cls					:       Dict		):
									super().tearDownClass()
									torch.use_deterministic_algorithms(_UpperCAmelCase		)







							def 							__lowercase			(							self					:       Dict		):
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()







							def 							__lowercase			(							self					:       List[Any]		):
									_a     :      List[str]        =							torch.manual_seed(51		)

									_a     :      str        =							StableDiffusionAttendAndExcitePipeline.from_pretrained(
									    'CompVis/stable-diffusion-v1-4'	,safety_checker=_UpperCAmelCase	,torch_dtype=torch.floataa		)
									pipe.to('cuda'		)

									_a     :      List[Any]        =							'a painting of an elephant with glasses'
									_a     :      Any        =							[5, 7]

									_a     :      List[Any]        =							pipe(
									    prompt=_UpperCAmelCase	,token_indices=_UpperCAmelCase	,guidance_scale=7.5	,generator=_UpperCAmelCase	,num_inference_steps=5	,max_iter_to_alter=5	,output_type='numpy'	,).images[0]

									_a     :      Any        =							load_numpy(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'		)
									assert np.abs((expected_image - image).max()		) < 5E-1









'''simple docstring'''

import multiprocessing
from typing import TYPE_CHECKING, Optional, Union

from .. import Dataset, Features, config
from ..formatting import query_table
from ..packaged_modules.sql.sql import Sql
from ..utils import logging
from .abc import AbstractDatasetInputStream


if TYPE_CHECKING:
							import sqlitea

							import sqlalchemy

class 	__magic_name__						(    _UpperCamelCase   ):







							def __init__(							self					:       Optional[int]	,_UpperCAmelCase					:       Union[str, "sqlalchemy.sql.Selectable"]	,_UpperCAmelCase					:       Union[str, "sqlalchemy.engine.Connection", "sqlalchemy.engine.Engine", "sqlite3.Connection"]	,_UpperCAmelCase					:       Optional[Features] = None	,_UpperCAmelCase					:       str = None	,_UpperCAmelCase					:       bool = False	,**_UpperCAmelCase					:       Dict	,):
									super().__init__(features=_UpperCAmelCase	,cache_dir=_UpperCAmelCase	,keep_in_memory=_UpperCAmelCase	,**_UpperCAmelCase		)
									_a     :      Tuple        =							Sql(
									    cache_dir=_UpperCAmelCase	,features=_UpperCAmelCase	,sql=_UpperCAmelCase	,con=_UpperCAmelCase	,**_UpperCAmelCase	,)







							def 							__lowercase			(							self					:       Dict		):
									_a     :      Optional[Any]        =							None
									_a     :      Dict        =							None
									_a     :      Dict        =							None
									_a     :      Optional[int]        =							None

									self.builder.download_and_prepare(
									    download_config=_UpperCAmelCase	,download_mode=_UpperCAmelCase	,verification_mode=_UpperCAmelCase	,base_path=_UpperCAmelCase	,)

									# Build dataset for splits
									_a     :      List[str]        =							self.builder.as_dataset(
									    split='train'	,verification_mode=_UpperCAmelCase	,in_memory=self.keep_in_memory		)
									return dataset

class 	__magic_name__						:







							def __init__(							self					:       Optional[int]	,_UpperCAmelCase					:       Dataset	,_UpperCAmelCase					:       str	,_UpperCAmelCase					:       Union[str, "sqlalchemy.engine.Connection", "sqlalchemy.engine.Engine", "sqlite3.Connection"]	,_UpperCAmelCase					:       Optional[int] = None	,_UpperCAmelCase					:       Optional[int] = None	,**_UpperCAmelCase					:       Dict	,):
									if num_proc is not None and num_proc <= 0:
											raise ValueError(F"""num_proc {num_proc} must be an integer > 0."""		)

									_a     :      Dict        =							dataset
									_a     :      List[Any]        =							name
									_a     :      Tuple        =							con
									_a     :      Union[str, Any]        =							batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
									_a     :      List[Any]        =							num_proc
									_a     :      Tuple        =							to_sql_kwargs







							def 							__lowercase			(							self					:       List[Any]		):
									_a     :      Tuple        =							self.to_sql_kwargs.pop('sql'	,_UpperCAmelCase		)
									_a     :      str        =							self.to_sql_kwargs.pop('con'	,_UpperCAmelCase		)
									_a     :      Optional[Any]        =							self.to_sql_kwargs.pop('index'	,_UpperCAmelCase		)

									_a     :      Any        =							self._write(index=_UpperCAmelCase	,**self.to_sql_kwargs		)
									return written







							def 							__lowercase			(							self					:       Optional[int]	,_UpperCAmelCase					:       Dict		):
									_a       ,					_a       ,					_a     :      Any        =							args
									_a     :      Tuple        =							{**to_sql_kwargs, 'if_exists': 'append'} if offset > 0 else to_sql_kwargs
									_a     :      Dict        =							query_table(
									    table=self.dataset.data	,key=slice(_UpperCAmelCase	,offset + self.batch_size		)	,indices=self.dataset._indices	,)
									_a     :      Tuple        =							batch.to_pandas()
									_a     :      Dict        =							df.to_sql(self.name	,self.con	,index=_UpperCAmelCase	,**_UpperCAmelCase		)
									return num_rows or len(_UpperCAmelCase		)







							def 							__lowercase			(							self					:       int	,_UpperCAmelCase					:       Optional[int]	,**_UpperCAmelCase					:       List[Any]		):
									_a     :      Union[str, Any]        =							0

									if self.num_proc is None or self.num_proc == 1:
											for offset in logging.tqdm(
											    range(0	,len(self.dataset		)	,self.batch_size		)	,unit='ba'	,disable=not logging.is_progress_bar_enabled()	,desc='Creating SQL from Arrow format'	,):
													written += self._batch_sql((offset, index, to_sql_kwargs)		)
									else:
											_a       ,					_a     :      List[Any]        =							len(self.dataset		), self.batch_size
											with multiprocessing.Pool(self.num_proc		) as pool:
													for num_rows in logging.tqdm(
													    pool.imap(
													        self._batch_sql	,[(offset, index, to_sql_kwargs) for offset in range(0	,_UpperCAmelCase	,_UpperCAmelCase		)]	,)	,total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size	,unit='ba'	,disable=not logging.is_progress_bar_enabled()	,desc='Creating SQL from Arrow format'	,):
															written += num_rows

									return written




