

import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


__lowercase     : Any       =						logging.get_logger(__name__)  # pylint: disable=invalid-name



class 						_A     (				snake_case				):





  '''simple docstring'''



  def __init__(  self   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,):

      '''simple docstring'''
      super().__init__()
      self.register_modules(
          vae=SCREAMING_SNAKE_CASE_   ,text_encoder=SCREAMING_SNAKE_CASE_   ,tokenizer=SCREAMING_SNAKE_CASE_   ,unet=SCREAMING_SNAKE_CASE_   ,scheduler=SCREAMING_SNAKE_CASE_   ,safety_checker=SCREAMING_SNAKE_CASE_   ,feature_extractor=SCREAMING_SNAKE_CASE_   ,)





  def    snake_case_       (  self   ,SCREAMING_SNAKE_CASE_ = "auto"    ):

      '''simple docstring'''
      if slice_size == "auto":
          # half the attention head size is usually a good trade-off between
          # speed and memory
          snake_case							:		List[str]        =		self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(SCREAMING_SNAKE_CASE_    )





  def    snake_case_       (  self    ):

      '''simple docstring'''
      self.enable_attention_slicing(SCREAMING_SNAKE_CASE_    )





  @torch.no_grad()
  def __call__(  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_ = None   ,**SCREAMING_SNAKE_CASE_   ,):

      '''simple docstring'''

      if isinstance(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    ):
          snake_case							:		List[Any]        =		1
      elif isinstance(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    ):
          snake_case							:		Any        =		len(SCREAMING_SNAKE_CASE_    )
      else:
          raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(SCREAMING_SNAKE_CASE_    )}"""    )

      if height % 8 != 0 or width % 8 != 0:
          raise ValueError(F"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""    )

      if (callback_steps is None) or (
          callback_steps is not None and (not isinstance(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    ) or callback_steps <= 0)
      ):
          raise ValueError(
              F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
              F""" {type(SCREAMING_SNAKE_CASE_    )}."""    )

      # get prompt text embeddings
      snake_case							:		Any        =		self.tokenizer(
          SCREAMING_SNAKE_CASE_   ,padding="""max_length"""   ,max_length=self.tokenizer.model_max_length   ,return_tensors="""pt"""   ,)
      snake_case							:		int        =		text_inputs.input_ids

      if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
          snake_case							:		str        =		self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]    )
          logger.warning(
              """The following part of your input was truncated because CLIP can only handle sequences up to"""
              F""" {self.tokenizer.model_max_length} tokens: {removed_text}"""    )
          snake_case							:		Optional[int]        =		text_input_ids[:, : self.tokenizer.model_max_length]

      if text_embeddings is None:
          snake_case							:		List[Any]        =		self.text_encoder(text_input_ids.to(self.device    )    )[0]

      # duplicate text embeddings for each generation per prompt, using mps friendly method
      snake_case      ,		snake_case      ,		snake_case							:		Optional[int]        =		text_embeddings.shape
      snake_case							:		Any        =		text_embeddings.repeat(1   ,SCREAMING_SNAKE_CASE_   ,1    )
      snake_case							:		List[str]        =		text_embeddings.view(bs_embed * num_images_per_prompt   ,SCREAMING_SNAKE_CASE_   ,-1    )

      # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
      # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
      # corresponds to doing no classifier free guidance.
      snake_case							:		int        =		guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
          snake_case							:		List[str]
          if negative_prompt is None:
              snake_case							:		int        =		[""""""]
          elif type(SCREAMING_SNAKE_CASE_    ) is not type(SCREAMING_SNAKE_CASE_    ):
              raise TypeError(
                  F"""`negative_prompt` should be the same type to `prompt`, but got {type(SCREAMING_SNAKE_CASE_    )} !="""
                  F""" {type(SCREAMING_SNAKE_CASE_    )}."""    )
          elif isinstance(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    ):
              snake_case							:		Dict        =		[negative_prompt]
          elif batch_size != len(SCREAMING_SNAKE_CASE_    ):
              raise ValueError(
                  F"""`negative_prompt`: {negative_prompt} has batch size {len(SCREAMING_SNAKE_CASE_    )}, but `prompt`:"""
                  F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                  """ the batch size of `prompt`."""    )
          else:
              snake_case							:		Optional[Any]        =		negative_prompt

          snake_case							:		List[str]        =		text_input_ids.shape[-1]
          snake_case							:		List[str]        =		self.tokenizer(
              SCREAMING_SNAKE_CASE_   ,padding="""max_length"""   ,max_length=SCREAMING_SNAKE_CASE_   ,truncation=SCREAMING_SNAKE_CASE_   ,return_tensors="""pt"""   ,)
          snake_case							:		str        =		self.text_encoder(uncond_input.input_ids.to(self.device    )    )[0]

          # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
          snake_case							:		Optional[Any]        =		uncond_embeddings.shape[1]
          snake_case							:		Union[str, Any]        =		uncond_embeddings.repeat(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,1    )
          snake_case							:		Any        =		uncond_embeddings.view(batch_size * num_images_per_prompt   ,SCREAMING_SNAKE_CASE_   ,-1    )

          # For classifier free guidance, we need to do two forward passes.
          # Here we concatenate the unconditional and text embeddings into a single batch
          # to avoid doing two forward passes
          snake_case							:		List[str]        =		torch.cat([uncond_embeddings, text_embeddings]    )

      # get the initial random noise unless the user supplied it

      # Unlike in other pipelines, latents need to be generated in the target device
      # for 1-to-1 results reproducibility with the CompVis implementation.
      # However this currently doesn't work in `mps`.
      snake_case							:		Union[str, Any]        =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
      snake_case							:		Tuple        =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
      snake_case							:		Optional[int]        =		text_embeddings.dtype
      if latents is None:
          if self.device.type == "mps":
              # randn does not exist on mps
              snake_case							:		Optional[int]        =		torch.randn(
                  SCREAMING_SNAKE_CASE_   ,generator=SCREAMING_SNAKE_CASE_   ,device="""cpu"""   ,dtype=SCREAMING_SNAKE_CASE_    ).to(self.device    )
              snake_case							:		Any        =		torch.randn(SCREAMING_SNAKE_CASE_   ,generator=SCREAMING_SNAKE_CASE_   ,device="""cpu"""   ,dtype=SCREAMING_SNAKE_CASE_    ).to(
                  self.device    )
          else:
              snake_case							:		List[Any]        =		torch.randn(
                  SCREAMING_SNAKE_CASE_   ,generator=SCREAMING_SNAKE_CASE_   ,device=self.device   ,dtype=SCREAMING_SNAKE_CASE_    )
              snake_case							:		Optional[int]        =		torch.randn(SCREAMING_SNAKE_CASE_   ,generator=SCREAMING_SNAKE_CASE_   ,device=self.device   ,dtype=SCREAMING_SNAKE_CASE_    )
      else:
          if latents_reference.shape != latents_shape:
              raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""    )
          snake_case							:		str        =		latents_reference.to(self.device    )
          snake_case							:		Optional[int]        =		latents.to(self.device    )

      # This is the key part of the pipeline where we
      # try to ensure that the generated images w/ the same seed
      # but different sizes actually result in similar images
      snake_case							:		Optional[Any]        =		(latents_shape[3] - latents_shape_reference[3]) // 2
      snake_case							:		List[Any]        =		(latents_shape[2] - latents_shape_reference[2]) // 2
      snake_case							:		List[Any]        =		latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
      snake_case							:		Dict        =		latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
      snake_case							:		List[str]        =		0 if dx < 0 else dx
      snake_case							:		Any        =		0 if dy < 0 else dy
      snake_case							:		Dict        =		max(-dx   ,0    )
      snake_case							:		Optional[Any]        =		max(-dy   ,0    )
      # import pdb
      # pdb.set_trace()
      snake_case							:		Optional[Any]        =		latents_reference[:, :, dy : dy + h, dx : dx + w]

      # set timesteps
      self.scheduler.set_timesteps(SCREAMING_SNAKE_CASE_    )

      # Some schedulers like PNDM have timesteps as arrays
      # It's more optimized to move all timesteps to correct device beforehand
      snake_case							:		Union[str, Any]        =		self.scheduler.timesteps.to(self.device    )

      # scale the initial noise by the standard deviation required by the scheduler
      snake_case							:		List[str]        =		latents * self.scheduler.init_noise_sigma

      # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
      # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
      # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
      # and should be between [0, 1]
      snake_case							:		List[str]        =		"""eta""" in set(inspect.signature(self.scheduler.step    ).parameters.keys()    )
      snake_case							:		Optional[Any]        =		{}
      if accepts_eta:
          snake_case							:		int        =		eta

      for i, t in enumerate(self.progress_bar(SCREAMING_SNAKE_CASE_    )    ):
          # expand the latents if we are doing classifier free guidance
          snake_case							:		Optional[Any]        =		torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents
          snake_case							:		List[str]        =		self.scheduler.scale_model_input(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    )

          # predict the noise residual
          snake_case							:		str        =		self.unet(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,encoder_hidden_states=SCREAMING_SNAKE_CASE_    ).sample

          # perform guidance
          if do_classifier_free_guidance:
              snake_case      ,		snake_case							:		str        =		noise_pred.chunk(2    )
              snake_case							:		List[Any]        =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

          # compute the previous noisy sample x_t -> x_t-1
          snake_case							:		int        =		self.scheduler.step(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,**SCREAMING_SNAKE_CASE_    ).prev_sample

          # call the callback, if provided
          if callback is not None and i % callback_steps == 0:
              callback(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    )

      snake_case							:		Optional[Any]        =		1 / 0.1_82_15 * latents
      snake_case							:		Any        =		self.vae.decode(SCREAMING_SNAKE_CASE_    ).sample

      snake_case							:		str        =		(image / 2 + 0.5).clamp(0   ,1    )

      # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
      snake_case							:		Optional[int]        =		image.cpu().permute(0   ,2   ,3   ,1    ).float().numpy()

      if self.safety_checker is not None:
          snake_case							:		Tuple        =		self.feature_extractor(self.numpy_to_pil(SCREAMING_SNAKE_CASE_    )   ,return_tensors="""pt"""    ).to(
              self.device    )
          snake_case      ,		snake_case							:		Optional[Any]        =		self.safety_checker(
              images=SCREAMING_SNAKE_CASE_   ,clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype    )    )
      else:
          snake_case							:		List[str]        =		None

      if output_type == "pil":
          snake_case							:		Tuple        =		self.numpy_to_pil(SCREAMING_SNAKE_CASE_    )

      if not return_dict:
          return (image, has_nsfw_concept)

      return StableDiffusionPipelineOutput(images=SCREAMING_SNAKE_CASE_   ,nsfw_content_detected=SCREAMING_SNAKE_CASE_    )





import unittest

import torch
from torch import nn

from accelerate.test_utils import require_cuda
from accelerate.utils.memory import find_executable_batch_size, release_memory

def 		UpperCamelCase						(   )		->						List[Any]:


  '''simple docstring'''






  raise RuntimeError("""CUDA out of memory.""" )



class    _snake_case (						nn.Module				):


 def __init__(       self):


   '''simple docstring'''




   super().__init__()
   lowercase__	: Optional[Any]     		=   nn.Linear(3       ,	4)
   lowercase__	: Union[str, Any]     		=   nn.BatchNormad(4)
   lowercase__	: str     		=   nn.Linear(4       ,	5)



 def   lowercase__  (       self       ,	SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   return self.lineara(self.batchnorm(self.lineara(SCREAMING_SNAKE_CASE_)))



class    _snake_case (						unittest.TestCase				):


 def   lowercase__  (       self):


   '''simple docstring'''




   lowercase__	: List[str]     		=   []

   @find_executable_batch_size(starting_batch_size=1_28)
   def mock_training_loop_function(SCREAMING_SNAKE_CASE_):
     nonlocal batch_sizes
     batch_sizes.append(SCREAMING_SNAKE_CASE_)
     if batch_size != 8:
       raise_fake_out_of_memory()

   mock_training_loop_function()
   self.assertListEqual(SCREAMING_SNAKE_CASE_       ,	[1_28, 64, 32, 16, 8])


 def   lowercase__  (       self):


   '''simple docstring'''




   lowercase__	: int     		=   []

   @find_executable_batch_size(starting_batch_size=1_28)
   def mock_training_loop_function(SCREAMING_SNAKE_CASE_       ,	SCREAMING_SNAKE_CASE_):
     nonlocal batch_sizes
     batch_sizes.append(SCREAMING_SNAKE_CASE_)
     if batch_size != 8:
       raise_fake_out_of_memory()
     return batch_size, arga

   lowercase__		,     lowercase__	: int     		=   mock_training_loop_function("""hello""")
   self.assertListEqual(SCREAMING_SNAKE_CASE_       ,	[1_28, 64, 32, 16, 8])
   self.assertListEqual([bs, arga]       ,	[8, """hello"""])


 def   lowercase__  (       self):


   '''simple docstring'''




   @find_executable_batch_size(starting_batch_size=0)
   def mock_training_loop_function(SCREAMING_SNAKE_CASE_):
     pass

   with self.assertRaises(SCREAMING_SNAKE_CASE_) as cm:
     mock_training_loop_function()
     self.assertIn("""No executable batch size found, reached zero."""       ,	cm.exception.args[0])


 def   lowercase__  (       self):


   '''simple docstring'''




   @find_executable_batch_size(starting_batch_size=16)
   def mock_training_loop_function(SCREAMING_SNAKE_CASE_):
     if batch_size > 0:
       raise_fake_out_of_memory()
     pass

   with self.assertRaises(SCREAMING_SNAKE_CASE_) as cm:
     mock_training_loop_function()
     self.assertIn("""No executable batch size found, reached zero."""       ,	cm.exception.args[0])


 def   lowercase__  (       self):


   '''simple docstring'''




   @find_executable_batch_size(starting_batch_size=1_28)
   def mock_training_loop_function(SCREAMING_SNAKE_CASE_       ,	SCREAMING_SNAKE_CASE_       ,	SCREAMING_SNAKE_CASE_):
     if batch_size != 8:
       raise raise_fake_out_of_memory()

   with self.assertRaises(SCREAMING_SNAKE_CASE_) as cm:
     mock_training_loop_function(1_28       ,	"""hello"""       ,	"""world""")
     self.assertIn("""Batch size was passed into `f`"""       ,	cm.exception.args[0])
     self.assertIn("""`f(arg1='hello', arg2='world')"""       ,	cm.exception.args[0])


 def   lowercase__  (       self):


   '''simple docstring'''




   @find_executable_batch_size(starting_batch_size=16)
   def mock_training_loop_function(SCREAMING_SNAKE_CASE_):
     raise ValueError("""Oops, we had an error!""")

   with self.assertRaises(SCREAMING_SNAKE_CASE_) as cm:
     mock_training_loop_function()
     self.assertIn("""Oops, we had an error!"""       ,	cm.exception.args[0])



 @require_cuda
 def   lowercase__  (       self):


   '''simple docstring'''




   lowercase__	: str     		=   torch.cuda.memory_allocated()
   lowercase__	: str     		=   ModelForTest()
   model.cuda()
   self.assertGreater(torch.cuda.memory_allocated()       ,	SCREAMING_SNAKE_CASE_)
   lowercase__	: Tuple     		=   release_memory(SCREAMING_SNAKE_CASE_)
   self.assertEqual(torch.cuda.memory_allocated()       ,	SCREAMING_SNAKE_CASE_)






