


"""simple docstring"""




from typing import Any, Callable, Dict, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker


__UpperCamelCase						=      '''CompVis/stable-diffusion-v1-1'''
__UpperCamelCase						=      '''CompVis/stable-diffusion-v1-2'''
__UpperCamelCase						=      '''CompVis/stable-diffusion-v1-3'''
__UpperCamelCase						=      '''CompVis/stable-diffusion-v1-4'''
class 					lowerCAmelCase	(					lowerCamelCase_    ):




		'''simple docstring'''







		def __init__(	self						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__ = True						,       )     ->							Optional[int]:
								super()._init_()

								SCREAMING_SNAKE_CASE													=					StableDiffusionPipeline.from_pretrained(lowerCAmelCase__				)
								SCREAMING_SNAKE_CASE													=					StableDiffusionPipeline.from_pretrained(lowerCAmelCase__				)
								SCREAMING_SNAKE_CASE													=					StableDiffusionPipeline.from_pretrained(lowerCAmelCase__				)
								SCREAMING_SNAKE_CASE													=					StableDiffusionPipeline(
								    vae=lowerCAmelCase__						,       text_encoder=lowerCAmelCase__						,       tokenizer=lowerCAmelCase__						,       unet=lowerCAmelCase__						,       scheduler=lowerCAmelCase__						,       safety_checker=lowerCAmelCase__						,       feature_extractor=lowerCAmelCase__						,       requires_safety_checker=lowerCAmelCase__						,       )

								self.register_modules(pipelinea=self.pipea						,       pipelinea=self.pipea						,       pipelinea=self.pipea						,       pipelinea=self.pipea				)







		@property
		def 							__A						(	self				)     ->							Dict[str, Any]:
								return {k: getattr(self						,       lowerCAmelCase__				) for k in self.config.keys() if not k.startswith('_'				)}







		def 							__A						(	self						,       lowerCAmelCase__ = "auto"				)     ->							List[Any]:
								if slice_size == "auto":
														# half the attention head size is usually a good trade-off between
														# speed and memory
														SCREAMING_SNAKE_CASE													=					self.unet.config.attention_head_dim // 2
								self.unet.set_attention_slice(lowerCAmelCase__				)







		def 							__A						(	self				)     ->							Dict:
								self.enable_attention_slicing(lowerCAmelCase__				)







		@torch.no_grad()
		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 50						,       lowerCAmelCase__ = 7.5						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       lowerCAmelCase__ = 0.0						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = "pil"						,       lowerCAmelCase__ = True						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       **lowerCAmelCase__						,       )     ->							Optional[Any]:
								return self.pipea(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )







		@torch.no_grad()
		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 50						,       lowerCAmelCase__ = 7.5						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       lowerCAmelCase__ = 0.0						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = "pil"						,       lowerCAmelCase__ = True						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       **lowerCAmelCase__						,       )     ->							Any:
								return self.pipea(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )







		@torch.no_grad()
		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 50						,       lowerCAmelCase__ = 7.5						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       lowerCAmelCase__ = 0.0						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = "pil"						,       lowerCAmelCase__ = True						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       **lowerCAmelCase__						,       )     ->							Optional[int]:
								return self.pipea(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )







		@torch.no_grad()
		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 50						,       lowerCAmelCase__ = 7.5						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       lowerCAmelCase__ = 0.0						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = "pil"						,       lowerCAmelCase__ = True						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       **lowerCAmelCase__						,       )     ->							int:
								return self.pipea(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )







		@torch.no_grad()
		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 512						,       lowerCAmelCase__ = 50						,       lowerCAmelCase__ = 7.5						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       lowerCAmelCase__ = 0.0						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = "pil"						,       lowerCAmelCase__ = True						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = 1						,       **lowerCAmelCase__						,       )     ->							List[Any]:
								SCREAMING_SNAKE_CASE													=					'cuda' if torch.cuda.is_available() else 'cpu'
								self.to(lowerCAmelCase__				)

								# Checks if the height and width are divisible by 8 or not
								if height % 8 != 0 or width % 8 != 0:
														raise ValueError(F'`height` and `width` must be divisible by 8 but are {height} and {width}.'				)

								# Get first result from Stable Diffusion Checkpoint v1.1
								SCREAMING_SNAKE_CASE													=					self.textaimg_sda_a(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )

								# Get first result from Stable Diffusion Checkpoint v1.2
								SCREAMING_SNAKE_CASE													=					self.textaimg_sda_a(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )

								# Get first result from Stable Diffusion Checkpoint v1.3
								SCREAMING_SNAKE_CASE													=					self.textaimg_sda_a(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )

								# Get first result from Stable Diffusion Checkpoint v1.4
								SCREAMING_SNAKE_CASE													=					self.textaimg_sda_a(
								    prompt=lowerCAmelCase__						,       height=lowerCAmelCase__						,       width=lowerCAmelCase__						,       num_inference_steps=lowerCAmelCase__						,       guidance_scale=lowerCAmelCase__						,       negative_prompt=lowerCAmelCase__						,       num_images_per_prompt=lowerCAmelCase__						,       eta=lowerCAmelCase__						,       generator=lowerCAmelCase__						,       latents=lowerCAmelCase__						,       output_type=lowerCAmelCase__						,       return_dict=lowerCAmelCase__						,       callback=lowerCAmelCase__						,       callback_steps=lowerCAmelCase__						,       **lowerCAmelCase__						,       )

								# Get all result images into a single list and pass it via StableDiffusionPipelineOutput for final result
								return StableDiffusionPipelineOutput([resa[0], resa[0], resa[0], resa[0]]				)






"""simple docstring"""




import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam
class 					lowerCAmelCase	(					datasets.BeamBasedBuilder    ):




		'''simple docstring'''







		def 							__A						(	self				)     ->							Dict:
								return datasets.DatasetInfo(
								    features=datasets.Features({'content': datasets.Value('string'				)}				)						,       supervised_keys=lowerCAmelCase__						,       )







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							str:
								return [datasets.SplitGenerator(name=datasets.Split.TRAIN						,       gen_kwargs={'examples': get_test_dummy_examples()}				)]







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Optional[int]:
								import apache_beam as beam

								return pipeline | "Load Examples" >> beam.Create(lowerCAmelCase__				)





class 					lowerCAmelCase	(					datasets.BeamBasedBuilder    ):




		'''simple docstring'''







		def 							__A						(	self				)     ->							int:
								return datasets.DatasetInfo(
								    features=datasets.Features({'a': datasets.Sequence({'b': datasets.Value('string'				)}				)}				)						,       supervised_keys=lowerCAmelCase__						,       )







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Dict:
								return [
								    datasets.SplitGenerator(name=datasets.Split.TRAIN						,       gen_kwargs={'examples': get_test_nested_examples()}				)
								]







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Optional[Any]:
								import apache_beam as beam

								return pipeline | "Load Examples" >> beam.Create(lowerCAmelCase__				)





def     lowercase			()   ->						str:
						return [(i, {"content": content}) for i, content in enumerate(['foo', 'bar', 'foobar']  )]


def     lowercase			()   ->						Optional[Any]:
						return [(i, {"a": {"b": [content]}}) for i, content in enumerate(['foo', 'bar', 'foobar']  )]


class 					lowerCAmelCase	(					lowerCamelCase_    ):




		'''simple docstring'''







		@require_beam
		def 							__A						(	self				)     ->							Optional[int]:
								SCREAMING_SNAKE_CASE													=					len(get_test_dummy_examples()				)
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					DummyBeamDataset(cache_dir=lowerCAmelCase__						,       beam_runner='DirectRunner'				)
														builder.download_and_prepare()
														self.assertTrue(
														    os.path.exists(
														        os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train.arrow'				)				)				)
														self.assertDictEqual(builder.info.features						,       datasets.Features({'content': datasets.Value('string'				)}				)				)
														SCREAMING_SNAKE_CASE													=					builder.as_dataset()
														self.assertEqual(dset['train'].num_rows						,       lowerCAmelCase__				)
														self.assertEqual(dset['train'].info.splits['train'].num_examples						,       lowerCAmelCase__				)
														self.assertDictEqual(dset['train'][0]						,       get_test_dummy_examples()[0][1]				)
														self.assertDictEqual(
														    dset['train'][expected_num_examples - 1]						,       get_test_dummy_examples()[expected_num_examples - 1][1]				)
														self.assertTrue(
														    os.path.exists(os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       'dataset_info.json'				)				)				)
														del dset







		@require_beam
		def 							__A						(	self				)     ->							int:
								import apache_beam as beam

								SCREAMING_SNAKE_CASE													=					beam.io.parquetio.WriteToParquet

								SCREAMING_SNAKE_CASE													=					len(get_test_dummy_examples()				)
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					DummyBeamDataset(cache_dir=lowerCAmelCase__						,       beam_runner='DirectRunner'				)
														with patch('apache_beam.io.parquetio.WriteToParquet'				) as write_parquet_mock:
																				SCREAMING_SNAKE_CASE													=					partial(lowerCAmelCase__						,       num_shards=2				)
																				builder.download_and_prepare()
														self.assertTrue(
														    os.path.exists(
														        os.path.join(
														            lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train-00000-of-00002.arrow'				)				)				)
														self.assertTrue(
														    os.path.exists(
														        os.path.join(
														            lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train-00000-of-00002.arrow'				)				)				)
														self.assertDictEqual(builder.info.features						,       datasets.Features({'content': datasets.Value('string'				)}				)				)
														SCREAMING_SNAKE_CASE													=					builder.as_dataset()
														self.assertEqual(dset['train'].num_rows						,       lowerCAmelCase__				)
														self.assertEqual(dset['train'].info.splits['train'].num_examples						,       lowerCAmelCase__				)
														# Order is not preserved when sharding, so we just check that all the elements are there
														self.assertListEqual(sorted(dset['train']['content']				)						,       sorted(['foo', 'bar', 'foobar']				)				)
														self.assertTrue(
														    os.path.exists(os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       'dataset_info.json'				)				)				)
														del dset







		@require_beam
		def 							__A						(	self				)     ->							Optional[Any]:
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					DummyBeamDataset(cache_dir=lowerCAmelCase__				)
														self.assertRaises(datasets.builder.MissingBeamOptions						,       builder.download_and_prepare				)







		@require_beam
		def 							__A						(	self				)     ->							int:
								SCREAMING_SNAKE_CASE													=					len(get_test_nested_examples()				)
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					NestedBeamDataset(cache_dir=lowerCAmelCase__						,       beam_runner='DirectRunner'				)
														builder.download_and_prepare()
														self.assertTrue(
														    os.path.exists(
														        os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train.arrow'				)				)				)
														self.assertDictEqual(
														    builder.info.features						,       datasets.Features({'a': datasets.Sequence({'b': datasets.Value('string'				)}				)}				)				)
														SCREAMING_SNAKE_CASE													=					builder.as_dataset()
														self.assertEqual(dset['train'].num_rows						,       lowerCAmelCase__				)
														self.assertEqual(dset['train'].info.splits['train'].num_examples						,       lowerCAmelCase__				)
														self.assertDictEqual(dset['train'][0]						,       get_test_nested_examples()[0][1]				)
														self.assertDictEqual(
														    dset['train'][expected_num_examples - 1]						,       get_test_nested_examples()[expected_num_examples - 1][1]				)
														self.assertTrue(
														    os.path.exists(os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       'dataset_info.json'				)				)				)
														del dset


