




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


lowerCamelCase__			:       int         =							'CompVis/stable-diffusion-v1-1'
lowerCamelCase__			:       Any         =							'CompVis/stable-diffusion-v1-2'
lowerCamelCase__			:       Optional[Any]         =							'CompVis/stable-diffusion-v1-3'
lowerCamelCase__			:       str         =							'CompVis/stable-diffusion-v1-4'





class 	lowerCamelCase_							(				_a ):


 '''simple docstring'''







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

       SCREAMING_SNAKE_CASE_ :    int     						=      StableDiffusionPipeline.from_pretrained(snake_case_						)
       SCREAMING_SNAKE_CASE_ :    Optional[int]     						=      StableDiffusionPipeline.from_pretrained(snake_case_						)
       SCREAMING_SNAKE_CASE_ :    Dict     						=      StableDiffusionPipeline.from_pretrained(snake_case_						)
       SCREAMING_SNAKE_CASE_ :    int     						=      StableDiffusionPipeline(
           vae=snake_case_    ,    text_encoder=snake_case_    ,    tokenizer=snake_case_    ,    unet=snake_case_    ,    scheduler=snake_case_    ,    safety_checker=snake_case_    ,    feature_extractor=snake_case_    ,    requires_safety_checker=snake_case_    ,    )

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



 @property
 def 				lowerCAmelCase_  (							self  :		Optional[int]						):
       return {k: getattr(self    ,    snake_case_						) for k in self.config.keys() if not k.startswith('_'						)}



 def 				lowerCAmelCase_  (							self  :		List[str]    ,    _lowerCAmelCase  :		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_ :    Dict     						=      self.unet.config.attention_head_dim // 2
       self.unet.set_attention_slice(snake_case_						)



 def 				lowerCAmelCase_  (							self  :		Optional[int]						):
       self.enable_attention_slicing(snake_case_						)



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



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



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



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




 @torch.no_grad()
 def 				lowerCAmelCase_  (							self  :		Optional[Any]    ,    _lowerCAmelCase  :		Union[str, List[str]]    ,    _lowerCAmelCase  :		int = 512    ,    _lowerCAmelCase  :		int = 512    ,    _lowerCAmelCase  :		int = 50    ,    _lowerCAmelCase  :		float = 7.5    ,    _lowerCAmelCase  :		Optional[Union[str, List[str]]] = None    ,    _lowerCAmelCase  :		Optional[int] = 1    ,    _lowerCAmelCase  :		float = 0.0    ,    _lowerCAmelCase  :		Optional[torch.Generator] = None    ,    _lowerCAmelCase  :		Optional[torch.FloatTensor] = None    ,    _lowerCAmelCase  :		Optional[str] = "pil"    ,    _lowerCAmelCase  :		bool = True    ,    _lowerCAmelCase  :		Optional[Callable[[int, int, torch.FloatTensor], None]] = None    ,    _lowerCAmelCase  :		int = 1    ,    **_lowerCAmelCase  :		Dict    ,    ):
       SCREAMING_SNAKE_CASE_ :    Union[str, Any]     						=      """cuda""" if torch.cuda.is_available() else """cpu"""
       self.to(snake_case_						)

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

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

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

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

       # 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]]						)





import gc
import threading
import time

import psutil
import torch





class 	lowerCamelCase_							:


 '''simple docstring'''







 def __init__(							self  :		Optional[Any]						):
       SCREAMING_SNAKE_CASE_     						=      psutil.Process()
       SCREAMING_SNAKE_CASE_     						=      False



 def 				lowerCAmelCase_  (							self  :		Dict						):
       SCREAMING_SNAKE_CASE_     						=      -1

       while True:
             SCREAMING_SNAKE_CASE_     						=      max(self.process.memory_info().rss    ,    self.cpu_memory_peak						)

             # can't sleep or will not catch the peak right (this comment is here on purpose)
             if not self.peak_monitoring:
                   break



 def 				lowerCAmelCase_  (							self  :		List[str]						):
       SCREAMING_SNAKE_CASE_     						=      True
       SCREAMING_SNAKE_CASE_     						=      threading.Thread(target=self.peak_monitor						)
       SCREAMING_SNAKE_CASE_     						=      True
       self.thread.start()




 def 				lowerCAmelCase_  (							self  :		List[str]						):
       SCREAMING_SNAKE_CASE_     						=      False
       self.thread.join()
       return self.cpu_memory_peak


lowerCamelCase__			:       List[str]         =							PeakCPUMemory()

def  UpperCAmelCase_							(				) ->			Tuple:
      # Time
      SCREAMING_SNAKE_CASE_     						=      {'time': time.time()}

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

      # CPU mem
      SCREAMING_SNAKE_CASE_     						=      psutil.Process().memory_info().rss
      cpu_peak_tracker.start()

      # GPU mem
      for i in range(torch.cuda.device_count()							):
            SCREAMING_SNAKE_CASE_     						=      torch.cuda.memory_allocated(__UpperCAmelCase							)
      torch.cuda.reset_peak_memory_stats()

      return measures

def  UpperCAmelCase_							(				__UpperCAmelCase						:     Optional[int]							) ->			Optional[Any]:
      # Time
      SCREAMING_SNAKE_CASE_     						=      {'time': time.time() - start_measures['time']}

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

      # CPU mem
      SCREAMING_SNAKE_CASE_     						=      (psutil.Process().memory_info().rss - start_measures['cpu']) / 2**20
      SCREAMING_SNAKE_CASE_     						=      (cpu_peak_tracker.stop() - start_measures['cpu']) / 2**20

      # GPU mem
      for i in range(torch.cuda.device_count()							):
            SCREAMING_SNAKE_CASE_     						=      (torch.cuda.memory_allocated(__UpperCAmelCase							) - start_measures[str(__UpperCAmelCase							)]) / 2**20
            SCREAMING_SNAKE_CASE_     						=      (torch.cuda.max_memory_allocated(__UpperCAmelCase							) - start_measures[str(__UpperCAmelCase							)]) / 2**20

      return measures





def  UpperCAmelCase_							(				__UpperCAmelCase						:     Optional[int]      ,      __UpperCAmelCase						:     Any							) ->			Optional[Any]:
      print(f"{description}:"							)
      print(f"- Time: {measures['time']:.2f}s"							)
      for i in range(torch.cuda.device_count()							):
            print(f"- GPU {i} allocated: {measures[str(__UpperCAmelCase							)]:.2f}MiB"							)
            SCREAMING_SNAKE_CASE_     						=      measures[f"{i}-peak"]
            print(f"- GPU {i} peak: {peak:.2f}MiB"							)
      print(f"- CPU RAM allocated: {measures['cpu']:.2f}MiB"							)
      print(f"- CPU RAM peak: {measures['cpu-peak']:.2f}MiB"							)