



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_						(    __A				):
       def __init__(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = True,     )	->						List[Any]:
          super()._init_()

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

          self.register_modules(pipelinea=self.pipea,     pipelinea=self.pipea,     pipelinea=self.pipea,     pipelinea=self.pipea			)
       @property
       def    snake_case_			(     self			)	->						Dict:
          return {k: getattr(self,     UpperCamelCase__			) for k in self.config.keys() if not k.startswith('_'			)}
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_ = "auto"			)	->						Dict:


          if slice_size == "auto":
             # half the attention head size is usually a good trade-off between
             # speed and memory
             UpperCamelCase       :    Optional[Any]		      =      self.unet.config.attention_head_dim // 2
          self.unet.set_attention_slice(UpperCamelCase__			)
       def    snake_case_			(     self			)	->						Union[str, Any]:
          self.enable_attention_slicing(UpperCamelCase__			)
       @torch.no_grad()
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 50,     SCREAMING_SNAKE_CASE_ = 7.5,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     SCREAMING_SNAKE_CASE_ = 0.0,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = "pil",     SCREAMING_SNAKE_CASE_ = True,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     **SCREAMING_SNAKE_CASE_,     )	->						Tuple:
          return self.pipea(
              prompt=UpperCamelCase__,     height=UpperCamelCase__,     width=UpperCamelCase__,     num_inference_steps=UpperCamelCase__,     guidance_scale=UpperCamelCase__,     negative_prompt=UpperCamelCase__,     num_images_per_prompt=UpperCamelCase__,     eta=UpperCamelCase__,     generator=UpperCamelCase__,     latents=UpperCamelCase__,     output_type=UpperCamelCase__,     return_dict=UpperCamelCase__,     callback=UpperCamelCase__,     callback_steps=UpperCamelCase__,     **UpperCamelCase__,     )
       @torch.no_grad()
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 50,     SCREAMING_SNAKE_CASE_ = 7.5,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     SCREAMING_SNAKE_CASE_ = 0.0,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = "pil",     SCREAMING_SNAKE_CASE_ = True,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     **SCREAMING_SNAKE_CASE_,     )	->						Optional[int]:
          return self.pipea(
              prompt=UpperCamelCase__,     height=UpperCamelCase__,     width=UpperCamelCase__,     num_inference_steps=UpperCamelCase__,     guidance_scale=UpperCamelCase__,     negative_prompt=UpperCamelCase__,     num_images_per_prompt=UpperCamelCase__,     eta=UpperCamelCase__,     generator=UpperCamelCase__,     latents=UpperCamelCase__,     output_type=UpperCamelCase__,     return_dict=UpperCamelCase__,     callback=UpperCamelCase__,     callback_steps=UpperCamelCase__,     **UpperCamelCase__,     )
       @torch.no_grad()
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 50,     SCREAMING_SNAKE_CASE_ = 7.5,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     SCREAMING_SNAKE_CASE_ = 0.0,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = "pil",     SCREAMING_SNAKE_CASE_ = True,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     **SCREAMING_SNAKE_CASE_,     )	->						Any:
          return self.pipea(
              prompt=UpperCamelCase__,     height=UpperCamelCase__,     width=UpperCamelCase__,     num_inference_steps=UpperCamelCase__,     guidance_scale=UpperCamelCase__,     negative_prompt=UpperCamelCase__,     num_images_per_prompt=UpperCamelCase__,     eta=UpperCamelCase__,     generator=UpperCamelCase__,     latents=UpperCamelCase__,     output_type=UpperCamelCase__,     return_dict=UpperCamelCase__,     callback=UpperCamelCase__,     callback_steps=UpperCamelCase__,     **UpperCamelCase__,     )
       @torch.no_grad()
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 50,     SCREAMING_SNAKE_CASE_ = 7.5,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     SCREAMING_SNAKE_CASE_ = 0.0,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = "pil",     SCREAMING_SNAKE_CASE_ = True,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     **SCREAMING_SNAKE_CASE_,     )	->						List[str]:
          return self.pipea(
              prompt=UpperCamelCase__,     height=UpperCamelCase__,     width=UpperCamelCase__,     num_inference_steps=UpperCamelCase__,     guidance_scale=UpperCamelCase__,     negative_prompt=UpperCamelCase__,     num_images_per_prompt=UpperCamelCase__,     eta=UpperCamelCase__,     generator=UpperCamelCase__,     latents=UpperCamelCase__,     output_type=UpperCamelCase__,     return_dict=UpperCamelCase__,     callback=UpperCamelCase__,     callback_steps=UpperCamelCase__,     **UpperCamelCase__,     )






       @torch.no_grad()
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 512,     SCREAMING_SNAKE_CASE_ = 50,     SCREAMING_SNAKE_CASE_ = 7.5,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     SCREAMING_SNAKE_CASE_ = 0.0,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = "pil",     SCREAMING_SNAKE_CASE_ = True,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = 1,     **SCREAMING_SNAKE_CASE_,     )	->						str:
          UpperCamelCase       :    Any		      =      'cuda' if torch.cuda.is_available() else 'cpu'
          self.to(UpperCamelCase__			)

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

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

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

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

          # 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 functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available


def       __UpperCamelCase					(      *lowercase_       :  Union[str, Any]   ):
       """simple docstring"""




       if not isinstance(lowercase_   ,       lowercase_   ):
              a_						     =			list(lowercase_   )
       for i in range(len(lowercase_   )   ):
              a_						     =			None
       gc.collect()
       if is_xpu_available():
              torch.xpu.empty_cache()
       elif is_npu_available():
              torch.npu.empty_cache()
       else:
              torch.cuda.empty_cache()
       return objects


def       __UpperCamelCase					(      lowercase_       :  Exception   ):
       """simple docstring"""




       a_						     =			[
           'CUDA out of memory.',  # CUDA OOM
           'cuDNN error: CUDNN_STATUS_NOT_SUPPORTED.',  # CUDNN SNAFU
           'DefaultCPUAllocator: can\'t allocate memory',  # CPU OOM
       ]
       if isinstance(lowercase_   ,       lowercase_   ) and len(exception.args   ) == 1:
              return any(err in exception.args[0] for err in _statements   )
       return False


def       __UpperCamelCase					(      lowercase_       :  callable = None   ,       lowercase_       :  int = 128   ):
       """simple docstring"""




       if function is None:
              return functools.partial(lowercase_   ,       starting_batch_size=lowercase_   )

       a_						     =			starting_batch_size

       def decorator(*lowercase_       :  str   ,       **lowercase_       :  Dict   ):
              nonlocal batch_size
              gc.collect()
              if is_xpu_available():
                     torch.xpu.empty_cache()
              elif is_npu_available():
                     torch.npu.empty_cache()
              else:
                     torch.cuda.empty_cache()
              a_						     =			list(inspect.signature(lowercase_   ).parameters.keys()   )
              # Guard against user error
              if len(lowercase_   ) < (len(lowercase_   ) + 1):
                     a_						     =			', '.join([F'{arg}={value}' for arg, value in zip(params[1:]   ,       args[1:]   )]   )
                     raise TypeError(
                         F'Batch size was passed into `{function.__name__}` as the first argument when called.'
                         F'Remove this as the decorator already does so: `{function.__name__}({arg_str})`'   )
              while True:
                     if batch_size == 0:
                            raise RuntimeError('No executable batch size found, reached zero.'   )
                     try:
                            return function(lowercase_   ,       *lowercase_   ,       **lowercase_   )
                     except Exception as e:
                            if should_reduce_batch_size(lowercase_   ):
                                   gc.collect()
                                   if is_xpu_available():
                                          torch.xpu.empty_cache()
                                   elif is_npu_available():
                                          torch.npu.empty_cache()
                                   else:
                                          torch.cuda.empty_cache()
                                   batch_size //= 2
                            else:
                                   raise

       return decorator




