


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


a_ :Dict  							=			'CompVis/stable-diffusion-v1-1'
a_ :str  							=			'CompVis/stable-diffusion-v1-2'
a_ :List[str]  							=			'CompVis/stable-diffusion-v1-3'
a_ :Dict  							=			'CompVis/stable-diffusion-v1-4'

class 							lowercase  (						_UpperCAmelCase			):







     def __init__(						self :						Any   ,   _lowercase :						AutoencoderKL   ,   _lowercase :						CLIPTextModel   ,   _lowercase :						CLIPTokenizer   ,   _lowercase :						UNetaDConditionModel   ,   _lowercase :						Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]   ,   _lowercase :						StableDiffusionSafetyChecker   ,   _lowercase :						CLIPImageProcessor   ,   _lowercase :						bool = True   ,   ):
           super()._init_()

           SCREAMING_SNAKE_CASE__       :  int						     =		StableDiffusionPipeline.from_pretrained(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		StableDiffusionPipeline.from_pretrained(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		StableDiffusionPipeline.from_pretrained(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		StableDiffusionPipeline(
               vae=_lowercase   ,   text_encoder=_lowercase   ,   tokenizer=_lowercase   ,   unet=_lowercase   ,   scheduler=_lowercase   ,   safety_checker=_lowercase   ,   feature_extractor=_lowercase   ,   requires_safety_checker=_lowercase   ,   )

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







     @property
     def 							lowercase__       (						self :						Any      ):
           return {k: getattr(self   ,   _lowercase      ) for k in self.config.keys() if not k.startswith('''_'''      )}







     def 							lowercase__       (						self :						Optional[Any]   ,   _lowercase :						Optional[Union[str, int]] = "auto"      ):
           if slice_size == "auto":
                 # half the attention head size is usually a good trade-off between
                 # speed and memory
                 SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.unet.config.attention_head_dim // 2
           self.unet.set_attention_slice(_lowercase      )







     def 							lowercase__       (						self :						List[str]      ):
           self.enable_attention_slicing(_lowercase      )







     @torch.no_grad()
     def 							lowercase__       (						self :						int   ,   _lowercase :						Union[str, List[str]]   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 50   ,   _lowercase :						float = 7.5   ,   _lowercase :						Optional[Union[str, List[str]]] = None   ,   _lowercase :						Optional[int] = 1   ,   _lowercase :						float = 0.0   ,   _lowercase :						Optional[torch.Generator] = None   ,   _lowercase :						Optional[torch.FloatTensor] = None   ,   _lowercase :						Optional[str] = "pil"   ,   _lowercase :						bool = True   ,   _lowercase :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,   _lowercase :						int = 1   ,   **_lowercase :						Optional[int]   ,   ):
           return self.pipea(
               prompt=_lowercase   ,   height=_lowercase   ,   width=_lowercase   ,   num_inference_steps=_lowercase   ,   guidance_scale=_lowercase   ,   negative_prompt=_lowercase   ,   num_images_per_prompt=_lowercase   ,   eta=_lowercase   ,   generator=_lowercase   ,   latents=_lowercase   ,   output_type=_lowercase   ,   return_dict=_lowercase   ,   callback=_lowercase   ,   callback_steps=_lowercase   ,   **_lowercase   ,   )







     @torch.no_grad()
     def 							lowercase__       (						self :						int   ,   _lowercase :						Union[str, List[str]]   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 50   ,   _lowercase :						float = 7.5   ,   _lowercase :						Optional[Union[str, List[str]]] = None   ,   _lowercase :						Optional[int] = 1   ,   _lowercase :						float = 0.0   ,   _lowercase :						Optional[torch.Generator] = None   ,   _lowercase :						Optional[torch.FloatTensor] = None   ,   _lowercase :						Optional[str] = "pil"   ,   _lowercase :						bool = True   ,   _lowercase :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,   _lowercase :						int = 1   ,   **_lowercase :						int   ,   ):
           return self.pipea(
               prompt=_lowercase   ,   height=_lowercase   ,   width=_lowercase   ,   num_inference_steps=_lowercase   ,   guidance_scale=_lowercase   ,   negative_prompt=_lowercase   ,   num_images_per_prompt=_lowercase   ,   eta=_lowercase   ,   generator=_lowercase   ,   latents=_lowercase   ,   output_type=_lowercase   ,   return_dict=_lowercase   ,   callback=_lowercase   ,   callback_steps=_lowercase   ,   **_lowercase   ,   )







     @torch.no_grad()
     def 							lowercase__       (						self :						Tuple   ,   _lowercase :						Union[str, List[str]]   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 50   ,   _lowercase :						float = 7.5   ,   _lowercase :						Optional[Union[str, List[str]]] = None   ,   _lowercase :						Optional[int] = 1   ,   _lowercase :						float = 0.0   ,   _lowercase :						Optional[torch.Generator] = None   ,   _lowercase :						Optional[torch.FloatTensor] = None   ,   _lowercase :						Optional[str] = "pil"   ,   _lowercase :						bool = True   ,   _lowercase :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,   _lowercase :						int = 1   ,   **_lowercase :						Union[str, Any]   ,   ):
           return self.pipea(
               prompt=_lowercase   ,   height=_lowercase   ,   width=_lowercase   ,   num_inference_steps=_lowercase   ,   guidance_scale=_lowercase   ,   negative_prompt=_lowercase   ,   num_images_per_prompt=_lowercase   ,   eta=_lowercase   ,   generator=_lowercase   ,   latents=_lowercase   ,   output_type=_lowercase   ,   return_dict=_lowercase   ,   callback=_lowercase   ,   callback_steps=_lowercase   ,   **_lowercase   ,   )







     @torch.no_grad()
     def 							lowercase__       (						self :						Optional[int]   ,   _lowercase :						Union[str, List[str]]   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 50   ,   _lowercase :						float = 7.5   ,   _lowercase :						Optional[Union[str, List[str]]] = None   ,   _lowercase :						Optional[int] = 1   ,   _lowercase :						float = 0.0   ,   _lowercase :						Optional[torch.Generator] = None   ,   _lowercase :						Optional[torch.FloatTensor] = None   ,   _lowercase :						Optional[str] = "pil"   ,   _lowercase :						bool = True   ,   _lowercase :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,   _lowercase :						int = 1   ,   **_lowercase :						Optional[Any]   ,   ):
           return self.pipea(
               prompt=_lowercase   ,   height=_lowercase   ,   width=_lowercase   ,   num_inference_steps=_lowercase   ,   guidance_scale=_lowercase   ,   negative_prompt=_lowercase   ,   num_images_per_prompt=_lowercase   ,   eta=_lowercase   ,   generator=_lowercase   ,   latents=_lowercase   ,   output_type=_lowercase   ,   return_dict=_lowercase   ,   callback=_lowercase   ,   callback_steps=_lowercase   ,   **_lowercase   ,   )







     @torch.no_grad()
     def 							lowercase__       (						self :						Optional[int]   ,   _lowercase :						Union[str, List[str]]   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 50   ,   _lowercase :						float = 7.5   ,   _lowercase :						Optional[Union[str, List[str]]] = None   ,   _lowercase :						Optional[int] = 1   ,   _lowercase :						float = 0.0   ,   _lowercase :						Optional[torch.Generator] = None   ,   _lowercase :						Optional[torch.FloatTensor] = None   ,   _lowercase :						Optional[str] = "pil"   ,   _lowercase :						bool = True   ,   _lowercase :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,   _lowercase :						int = 1   ,   **_lowercase :						int   ,   ):
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		'''cuda''' if torch.cuda.is_available() else '''cpu'''
           self.to(_lowercase      )

           # 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__       :  Any						     =		self.textaimg_sda_a(
               prompt=_lowercase   ,   height=_lowercase   ,   width=_lowercase   ,   num_inference_steps=_lowercase   ,   guidance_scale=_lowercase   ,   negative_prompt=_lowercase   ,   num_images_per_prompt=_lowercase   ,   eta=_lowercase   ,   generator=_lowercase   ,   latents=_lowercase   ,   output_type=_lowercase   ,   return_dict=_lowercase   ,   callback=_lowercase   ,   callback_steps=_lowercase   ,   **_lowercase   ,   )

           # Get first result from Stable Diffusion Checkpoint v1.2
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		self.textaimg_sda_a(
               prompt=_lowercase   ,   height=_lowercase   ,   width=_lowercase   ,   num_inference_steps=_lowercase   ,   guidance_scale=_lowercase   ,   negative_prompt=_lowercase   ,   num_images_per_prompt=_lowercase   ,   eta=_lowercase   ,   generator=_lowercase   ,   latents=_lowercase   ,   output_type=_lowercase   ,   return_dict=_lowercase   ,   callback=_lowercase   ,   callback_steps=_lowercase   ,   **_lowercase   ,   )

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

           # Get first result from Stable Diffusion Checkpoint v1.4
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		self.textaimg_sda_a(
               prompt=_lowercase   ,   height=_lowercase   ,   width=_lowercase   ,   num_inference_steps=_lowercase   ,   guidance_scale=_lowercase   ,   negative_prompt=_lowercase   ,   num_images_per_prompt=_lowercase   ,   eta=_lowercase   ,   generator=_lowercase   ,   latents=_lowercase   ,   output_type=_lowercase   ,   return_dict=_lowercase   ,   callback=_lowercase   ,   callback_steps=_lowercase   ,   **_lowercase   ,   )

           # 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 pytest

from datasets.splits import SplitDict, SplitInfo
from datasets.utils.py_utils import asdict


@pytest.mark.parametrize(
    '''split_dict''',   [
        SplitDict(),
        SplitDict({'''train''': SplitInfo(name='''train''',   num_bytes=1_337,   num_examples=42,   dataset_name='''my_dataset'''						)}						),
        SplitDict({'''train''': SplitInfo(name='''train''',   num_bytes=1_337,   num_examples=42						)}						),
        SplitDict({'''train''': SplitInfo()}						),
    ],   )
def        a__     (					UpperCamelCase_						:      SplitDict						):
   UpperCAmelCase__	:List[Any]      =      split_dict._to_yaml_list()
   assert len(UpperCamelCase_						) == len(UpperCamelCase_						)
   UpperCAmelCase__	:Tuple      =      SplitDict._from_yaml_list(UpperCamelCase_						)
   for split_name, split_info in split_dict.items():
      # dataset_name field is deprecated, and is therefore not part of the YAML dump
      UpperCAmelCase__	:Union[str, Any]      =      None
      # the split name of split_dict takes over the name of the split info object
      UpperCAmelCase__	:Union[str, Any]      =      split_name
   assert split_dict == reloaded


@pytest.mark.parametrize(
    '''split_info''',   [SplitInfo(), SplitInfo(dataset_name=UpperCamelCase_						), SplitInfo(dataset_name='''my_dataset'''						)]						)
def        a__     (					UpperCamelCase_						:      List[Any]						):
   # For backward compatibility, we need asdict(split_dict) to return split info dictrionaries with the "dataset_name"
   # field even if it's deprecated. This way old versionso of `datasets` can still reload dataset_infos.json files
   UpperCAmelCase__	:List[Any]      =      asdict(SplitDict({'''train''': split_info}						)						)
   assert "dataset_name" in split_dict_asdict["train"]
   assert split_dict_asdict["train"]["dataset_name"] == split_info.dataset_name



