
'''simple docstring'''







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 				lowerCAmelCase__     (						lowerCamelCase_					,		lowerCamelCase_					,		unittest.TestCase  ):
					lowerCAmelCase_       =  StableDiffusionParadigmsPipeline
					lowerCAmelCase_       =  TEXT_TO_IMAGE_PARAMS
					lowerCAmelCase_       =  TEXT_TO_IMAGE_BATCH_PARAMS
					lowerCAmelCase_       =  TEXT_TO_IMAGE_IMAGE_PARAMS
					lowerCAmelCase_       =  TEXT_TO_IMAGE_IMAGE_PARAMS






					def 			_snake_case					(     self	):


											"""simple docstring"""





											torch.manual_seed(0	)
											lowercase_			:  Optional[Any]									= 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       ,							)
											lowercase_			:  Optional[int]									= DDIMParallelScheduler(
											    beta_start=0.00_085       ,							beta_end=0.012       ,							beta_schedule='''scaled_linear'''       ,							clip_sample=__SCREAMING_SNAKE_CASE       ,							set_alpha_to_one=__SCREAMING_SNAKE_CASE       ,							)
											torch.manual_seed(0	)
											lowercase_			:  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=1_28       ,							)
											torch.manual_seed(0	)
											lowercase_			:  List[str]									= 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=10_00       ,							hidden_act='''gelu'''       ,							projection_dim=5_12       ,							)
											lowercase_			:  int									= CLIPTextModel(__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Union[str, Any]									= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''	)

											lowercase_			:  Union[str, Any]									= {
											    '''unet''': unet,
											    '''scheduler''': scheduler,
											    '''vae''': vae,
											    '''text_encoder''': text_encoder,
											    '''tokenizer''': tokenizer,
											    '''safety_checker''': None,
											    '''feature_extractor''': None,
											}
											return components






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE=0	):


											"""simple docstring"""





											if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
																	lowercase_			:  Dict									= torch.manual_seed(__SCREAMING_SNAKE_CASE	)
											else:
																	lowercase_			:  List[Any]									= torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
											lowercase_			:  List[Any]									= {
											    '''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 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  Tuple									= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
											lowercase_			:  Optional[Any]									= self.get_dummy_components()
											lowercase_			:  List[Any]									= StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Dict									= sd_pipe.to(__SCREAMING_SNAKE_CASE	)
											sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

											lowercase_			:  int									= self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Dict									= sd_pipe(**__SCREAMING_SNAKE_CASE	).images
											lowercase_			:  str									= image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											lowercase_			:  Optional[int]									= np.array([0.4_773, 0.5_417, 0.4_723, 0.4_925, 0.5_631, 0.4_752, 0.5_240, 0.4_935, 0.5_023]	)

											assert np.abs(image_slice.flatten() - expected_slice	).max() < 1E-2






					def 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  Union[str, Any]									= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
											lowercase_			:  Any									= self.get_dummy_components()
											torch.manual_seed(0	)
											lowercase_			:  int									= DDPMParallelScheduler()
											torch.manual_seed(0	)
											lowercase_			:  Optional[Any]									= StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Optional[int]									= sd_pipe.to(__SCREAMING_SNAKE_CASE	)
											sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

											lowercase_			:  Union[str, Any]									= self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
											lowercase_			:  str									= sd_pipe(**__SCREAMING_SNAKE_CASE	).images
											lowercase_			:  str									= image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											lowercase_			:  Union[str, Any]									= np.array([0.3_573, 0.4_420, 0.4_960, 0.4_799, 0.3_796, 0.3_879, 0.4_819, 0.4_365, 0.4_468]	)

											assert np.abs(image_slice.flatten() - expected_slice	).max() < 1E-2






					def 			_snake_case					(     self	):


											"""simple docstring"""





											super().test_inference_batch_consistent(batch_sizes=[1, 2]	)






					def 			_snake_case					(     self	):


											"""simple docstring"""





											super().test_inference_batch_single_identical(batch_size=2       ,							expected_max_diff=3E-3	)






					def 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  Optional[int]									= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
											lowercase_			:  Optional[Any]									= self.get_dummy_components()
											lowercase_			:  Dict									= StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Any									= sd_pipe.to(__SCREAMING_SNAKE_CASE	)
											sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

											lowercase_			:  List[Any]									= self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
											lowercase_			:  int									= '''french fries'''
											lowercase_			:  str									= sd_pipe(**__SCREAMING_SNAKE_CASE       ,							negative_prompt=__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Optional[int]									= output.images
											lowercase_			:  Optional[int]									= image[0, -3:, -3:, -1]

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

											lowercase_			:  Union[str, Any]									= np.array([0.4_771, 0.5_420, 0.4_683, 0.4_918, 0.5_636, 0.4_725, 0.5_230, 0.4_923, 0.5_015]	)

											assert np.abs(image_slice.flatten() - expected_slice	).max() < 1E-2







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






					def 			_snake_case					(     self	):


											"""simple docstring"""





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






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE=0	):


											"""simple docstring"""





											lowercase_			:  List[Any]									= torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
											lowercase_			:  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 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  str									= '''stabilityai/stable-diffusion-2-base'''
											lowercase_			:  Optional[int]									= DDIMParallelScheduler.from_pretrained(__SCREAMING_SNAKE_CASE       ,							subfolder='''scheduler'''	)
											lowercase_			:  List[str]									= 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()

											lowercase_			:  Union[str, Any]									= self.get_inputs()
											lowercase_			:  Tuple									= pipe(**__SCREAMING_SNAKE_CASE	).images
											lowercase_			:  int									= image[0, -3:, -3:, -1].flatten()

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

											lowercase_			:  Union[str, Any]									= np.array([0.9_622, 0.9_602, 0.9_748, 0.9_591, 0.9_630, 0.9_691, 0.9_661, 0.9_631, 0.9_741]	)

											assert np.abs(expected_slice - image_slice	).max() < 1E-2





'''simple docstring'''







import multiprocessing
import os
from typing import BinaryIO, Optional, Union

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.csv.csv import Csv
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader






class 				lowerCAmelCase__     (						lowerCamelCase_  ):






					def __init__(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = False       ,							__SCREAMING_SNAKE_CASE = False       ,							__SCREAMING_SNAKE_CASE = None       ,							**__SCREAMING_SNAKE_CASE       ,							):


											"""simple docstring"""





											super().__init__(
											    __SCREAMING_SNAKE_CASE       ,							split=__SCREAMING_SNAKE_CASE       ,							features=__SCREAMING_SNAKE_CASE       ,							cache_dir=__SCREAMING_SNAKE_CASE       ,							keep_in_memory=__SCREAMING_SNAKE_CASE       ,							streaming=__SCREAMING_SNAKE_CASE       ,							num_proc=__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE       ,							)
											lowercase_			:  int									= path_or_paths if isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	) else {self.split: path_or_paths}
											lowercase_			:  List[Any]									= Csv(
											    cache_dir=__SCREAMING_SNAKE_CASE       ,							data_files=__SCREAMING_SNAKE_CASE       ,							features=__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE       ,							)






					def 			_snake_case					(     self	):


											"""simple docstring"""





											if self.streaming:
																	lowercase_			:  List[str]									= self.builder.as_streaming_dataset(split=self.split	)
											# Build regular (map-style) dataset
											else:
																	lowercase_			:  Any									= None
																	lowercase_			:  Optional[Any]									= None
																	lowercase_			:  str									= None
																	lowercase_			:  Optional[int]									= None

																	self.builder.download_and_prepare(
																	    download_config=__SCREAMING_SNAKE_CASE       ,							download_mode=__SCREAMING_SNAKE_CASE       ,							verification_mode=__SCREAMING_SNAKE_CASE       ,							base_path=__SCREAMING_SNAKE_CASE       ,							num_proc=self.num_proc       ,							)
																	lowercase_			:  Tuple									= self.builder.as_dataset(
																	    split=self.split       ,							verification_mode=__SCREAMING_SNAKE_CASE       ,							in_memory=self.keep_in_memory	)
											return dataset







class 				lowerCAmelCase__     :






					def __init__(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = None       ,							**__SCREAMING_SNAKE_CASE       ,							):


											"""simple docstring"""





											if num_proc is not None and num_proc <= 0:
																	raise ValueError(F'''num_proc {num_proc} must be an integer > 0.'''	)

											lowercase_			:  List[Any]									= dataset
											lowercase_			:  Dict									= path_or_buf
											lowercase_			:  int									= batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
											lowercase_			:  int									= num_proc
											lowercase_			:  Optional[int]									= '''utf-8'''
											lowercase_			:  Optional[Any]									= to_csv_kwargs






					def 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  List[str]									= self.to_csv_kwargs.pop('''path_or_buf'''       ,							__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Optional[Any]									= self.to_csv_kwargs.pop('''header'''       ,							__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Dict									= self.to_csv_kwargs.pop('''index'''       ,							__SCREAMING_SNAKE_CASE	)

											if isinstance(self.path_or_buf       ,							(str, bytes, os.PathLike)	):
																	with open(self.path_or_buf       ,							'''wb+'''	) as buffer:
																							lowercase_			:  Any									= self._write(file_obj=__SCREAMING_SNAKE_CASE       ,							header=__SCREAMING_SNAKE_CASE       ,							index=__SCREAMING_SNAKE_CASE       ,							**self.to_csv_kwargs	)
											else:
																	lowercase_			:  str									= self._write(file_obj=self.path_or_buf       ,							header=__SCREAMING_SNAKE_CASE       ,							index=__SCREAMING_SNAKE_CASE       ,							**self.to_csv_kwargs	)
											return written






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											lowercase_		,				lowercase_		,				lowercase_		,				lowercase_			:  int									= args

											lowercase_			:  List[str]									= query_table(
											    table=self.dataset.data       ,							key=slice(__SCREAMING_SNAKE_CASE       ,							offset + self.batch_size	)       ,							indices=self.dataset._indices       ,							)
											lowercase_			:  Dict									= batch.to_pandas().to_csv(
											    path_or_buf=__SCREAMING_SNAKE_CASE       ,							header=header if (offset == 0) else False       ,							index=__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	)
											return csv_str.encode(self.encoding	)






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											lowercase_			:  Dict									= 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 CSV from Arrow format'''       ,							):
																							lowercase_			:  Any									= self._batch_csv((offset, header, index, to_csv_kwargs)	)
																							written += file_obj.write(__SCREAMING_SNAKE_CASE	)

											else:
																	lowercase_		,				lowercase_			:  Union[str, Any]									= len(self.dataset	), self.batch_size
																	with multiprocessing.Pool(self.num_proc	) as pool:
																							for csv_str in logging.tqdm(
																							    pool.imap(
																							        self._batch_csv       ,							[(offset, header, index, to_csv_kwargs) for offset in range(0       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	)]       ,							)       ,							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 CSV from Arrow format'''       ,							):
																													written += file_obj.write(__SCREAMING_SNAKE_CASE	)

											return written



