


'''simple docstring'''



import inspect
from typing import List, Optional, Union

import torch
from torch import nn
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPImageProcessor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput



class       a							(						nn.Module			):


     """simple docstring"""






     def __init__(							self	:						List[str]				,					snake_case	:						int				,					snake_case	:						Dict=1.0     )		->				str:
          super().__init__()

          __UpperCAmelCase      :					Any    				=     cut_size
          __UpperCAmelCase      :					Any    				=     cut_power






     def 					lowerCamelCase__					(							self	:						Tuple				,					snake_case	:						str				,					snake_case	:						Tuple     )		->				List[Any]:
          __UpperCAmelCase				,  __UpperCAmelCase      :					Union[str, Any]    				=     pixel_values.shape[2:4]
          __UpperCAmelCase      :					Tuple    				=     min(snake_case				,					snake_case     )
          __UpperCAmelCase      :					Optional[int]    				=     min(snake_case				,					snake_case				,					self.cut_size     )
          __UpperCAmelCase      :					int    				=     []
          for _ in range(snake_case     ):
               __UpperCAmelCase      :					Optional[Any]    				=     int(torch.rand([]     ) ** self.cut_power * (max_size - min_size) + min_size     )
               __UpperCAmelCase      :					str    				=     torch.randint(0				,					sideX - size + 1				,					()     )
               __UpperCAmelCase      :					Optional[int]    				=     torch.randint(0				,					sideY - size + 1				,					()     )
               __UpperCAmelCase      :					List[Any]    				=     pixel_values[:, :, offsety : offsety + size, offsetx : offsetx + size]
               cutouts.append(F.adaptive_avg_poolad(snake_case				,					self.cut_size     )     )
          return torch.cat(snake_case     )




def       _a      (		_lowercase   :      Union[str, Any]     ,   _lowercase   :      int    ):





     '''simple docstring'''






     __UpperCAmelCase      :					List[Any]    				=     F.normalize(_lowercase     ,   dim=-1    )
     __UpperCAmelCase      :					List[str]    				=     F.normalize(_lowercase     ,   dim=-1    )
     return (x - y).norm(dim=-1    ).div(2    ).arcsin().pow(2    ).mul(2    )




def       _a      (		_lowercase   :      Union[str, Any]     ,   _lowercase   :      Any    ):





     '''simple docstring'''






     for param in model.parameters():
          __UpperCAmelCase      :					Tuple    				=     value



class       a							(						_a			):


     """simple docstring"""






     def __init__(							self	:						Optional[int]				,					snake_case	:						AutoencoderKL				,					snake_case	:						CLIPTextModel				,					snake_case	:						CLIPModel				,					snake_case	:						CLIPTokenizer				,					snake_case	:						UNetaDConditionModel				,					snake_case	:						Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]				,					snake_case	:						CLIPImageProcessor				,					)		->				Dict:
          super().__init__()
          self.register_modules(
              vae=snake_case				,					text_encoder=snake_case				,					clip_model=snake_case				,					tokenizer=snake_case				,					unet=snake_case				,					scheduler=snake_case				,					feature_extractor=snake_case				,					)

          __UpperCAmelCase      :					Any    				=     transforms.Normalize(mean=feature_extractor.image_mean				,					std=feature_extractor.image_std     )
          __UpperCAmelCase      :					Dict    				=     (
              feature_extractor.size
              if isinstance(feature_extractor.size				,					snake_case     )
              else feature_extractor.size['''shortest_edge''']
          )
          __UpperCAmelCase      :					Optional[int]    				=     MakeCutouts(self.cut_out_size     )

          set_requires_grad(self.text_encoder				,					snake_case     )
          set_requires_grad(self.clip_model				,					snake_case     )

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

     def 					lowerCamelCase__					(							self	:						Optional[Any]     )		->				Any:
          self.enable_attention_slicing(snake_case     )

     def 					lowerCamelCase__					(							self	:						Any     )		->				Dict:
          set_requires_grad(self.vae				,					snake_case     )

     def 					lowerCamelCase__					(							self	:						Optional[Any]     )		->				List[str]:
          set_requires_grad(self.vae				,					snake_case     )

     def 					lowerCamelCase__					(							self	:						str     )		->				Tuple:
          set_requires_grad(self.unet				,					snake_case     )

     def 					lowerCamelCase__					(							self	:						str     )		->				List[Any]:
          set_requires_grad(self.unet				,					snake_case     )

     @torch.enable_grad()
     def 					lowerCamelCase__					(							self	:						Optional[Any]				,					snake_case	:						List[Any]				,					snake_case	:						str				,					snake_case	:						Tuple				,					snake_case	:						List[Any]				,					snake_case	:						Optional[int]				,					snake_case	:						List[str]				,					snake_case	:						Union[str, Any]				,					snake_case	:						Any				,					snake_case	:						List[str]=True				,					)		->				Dict:
          __UpperCAmelCase      :					List[str]    				=     latents.detach().requires_grad_()

          __UpperCAmelCase      :					int    				=     self.scheduler.scale_model_input(snake_case				,					snake_case     )

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

          if isinstance(self.scheduler				,					(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)     ):
               __UpperCAmelCase      :					List[Any]    				=     self.scheduler.alphas_cumprod[timestep]
               __UpperCAmelCase      :					Tuple    				=     1 - alpha_prod_t
               # compute predicted original sample from predicted noise also called
               # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
               __UpperCAmelCase      :					List[str]    				=     (latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

               __UpperCAmelCase      :					List[Any]    				=     torch.sqrt(snake_case     )
               __UpperCAmelCase      :					Optional[int]    				=     pred_original_sample * (fac) + latents * (1 - fac)
          elif isinstance(self.scheduler				,					snake_case     ):
               __UpperCAmelCase      :					Union[str, Any]    				=     self.scheduler.sigmas[index]
               __UpperCAmelCase      :					Dict    				=     latents - sigma * noise_pred
          else:
               raise ValueError(f'scheduler type {type(self.scheduler     )} not supported'     )

          __UpperCAmelCase      :					Optional[int]    				=     1 / self.vae.config.scaling_factor * sample
          __UpperCAmelCase      :					List[Any]    				=     self.vae.decode(snake_case     ).sample
          __UpperCAmelCase      :					List[str]    				=     (image / 2 + 0.5).clamp(0				,					1     )

          if use_cutouts:
               __UpperCAmelCase      :					List[str]    				=     self.make_cutouts(snake_case				,					snake_case     )
          else:
               __UpperCAmelCase      :					Union[str, Any]    				=     transforms.Resize(self.cut_out_size     )(snake_case     )
          __UpperCAmelCase      :					Optional[Any]    				=     self.normalize(snake_case     ).to(latents.dtype     )

          __UpperCAmelCase      :					List[Any]    				=     self.clip_model.get_image_features(snake_case     )
          __UpperCAmelCase      :					Optional[int]    				=     image_embeddings_clip / image_embeddings_clip.norm(p=2				,					dim=-1				,					keepdim=snake_case     )

          if use_cutouts:
               __UpperCAmelCase      :					int    				=     spherical_dist_loss(snake_case				,					snake_case     )
               __UpperCAmelCase      :					Optional[Any]    				=     dists.view([num_cutouts, sample.shape[0], -1]     )
               __UpperCAmelCase      :					Dict    				=     dists.sum(2     ).mean(0     ).sum() * clip_guidance_scale
          else:
               __UpperCAmelCase      :					List[Any]    				=     spherical_dist_loss(snake_case				,					snake_case     ).mean() * clip_guidance_scale

          __UpperCAmelCase      :					Dict    				=     -torch.autograd.grad(snake_case				,					snake_case     )[0]

          if isinstance(self.scheduler				,					snake_case     ):
               __UpperCAmelCase      :					str    				=     latents.detach() + grads * (sigma**2)
               __UpperCAmelCase      :					Union[str, Any]    				=     noise_pred_original
          else:
               __UpperCAmelCase      :					List[Any]    				=     noise_pred_original - torch.sqrt(snake_case     ) * grads
          return noise_pred, latents






     @torch.no_grad()
     def __call__(							self	:						Optional[int]				,					snake_case	:						Union[str, List[str]]				,					snake_case	:						Optional[int] = 512				,					snake_case	:						Optional[int] = 512				,					snake_case	:						Optional[int] = 50				,					snake_case	:						Optional[float] = 7.5				,					snake_case	:						Optional[int] = 1				,					snake_case	:						float = 0.0				,					snake_case	:						Optional[float] = 100				,					snake_case	:						Optional[Union[str, List[str]]] = None				,					snake_case	:						Optional[int] = 4				,					snake_case	:						Optional[bool] = True				,					snake_case	:						Optional[torch.Generator] = None				,					snake_case	:						Optional[torch.FloatTensor] = None				,					snake_case	:						Optional[str] = "pil"				,					snake_case	:						bool = True				,					)		->				Dict:
          if isinstance(snake_case				,					snake_case     ):
               __UpperCAmelCase      :					Optional[Any]    				=     1
          elif isinstance(snake_case				,					snake_case     ):
               __UpperCAmelCase      :					Dict    				=     len(snake_case     )
          else:
               raise ValueError(f'`prompt` has to be of type `str` or `list` but is {type(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}.'     )

          # get prompt text embeddings
          __UpperCAmelCase      :					Optional[int]    				=     self.tokenizer(
              snake_case				,					padding='''max_length'''				,					max_length=self.tokenizer.model_max_length				,					truncation=snake_case				,					return_tensors='''pt'''				,					)
          __UpperCAmelCase      :					List[Any]    				=     self.text_encoder(text_input.input_ids.to(self.device     )     )[0]
          # duplicate text embeddings for each generation per prompt
          __UpperCAmelCase      :					int    				=     text_embeddings.repeat_interleave(snake_case				,					dim=0     )

          if clip_guidance_scale > 0:
               if clip_prompt is not None:
                    __UpperCAmelCase      :					str    				=     self.tokenizer(
                        snake_case				,					padding='''max_length'''				,					max_length=self.tokenizer.model_max_length				,					truncation=snake_case				,					return_tensors='''pt'''				,					).input_ids.to(self.device     )
               else:
                    __UpperCAmelCase      :					Dict    				=     text_input.input_ids.to(self.device     )
               __UpperCAmelCase      :					List[str]    				=     self.clip_model.get_text_features(snake_case     )
               __UpperCAmelCase      :					Any    				=     text_embeddings_clip / text_embeddings_clip.norm(p=2				,					dim=-1				,					keepdim=snake_case     )
               # duplicate text embeddings clip for each generation per prompt
               __UpperCAmelCase      :					List[str]    				=     text_embeddings_clip.repeat_interleave(snake_case				,					dim=0     )

          # 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.
          __UpperCAmelCase      :					List[Any]    				=     guidance_scale > 1.0
          # get unconditional embeddings for classifier free guidance
          if do_classifier_free_guidance:
               __UpperCAmelCase      :					Any    				=     text_input.input_ids.shape[-1]
               __UpperCAmelCase      :					Union[str, Any]    				=     self.tokenizer(['''''']				,					padding='''max_length'''				,					max_length=snake_case				,					return_tensors='''pt'''     )
               __UpperCAmelCase      :					List[str]    				=     self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]
               # duplicate unconditional embeddings for each generation per prompt
               __UpperCAmelCase      :					Optional[Any]    				=     uncond_embeddings.repeat_interleave(snake_case				,					dim=0     )

               # 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
               __UpperCAmelCase      :					Any    				=     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`.
          __UpperCAmelCase      :					str    				=     (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
          __UpperCAmelCase      :					List[str]    				=     text_embeddings.dtype
          if latents is None:
               if self.device.type == "mps":
                    # randn does not work reproducibly on mps
                    __UpperCAmelCase      :					Optional[Any]    				=     torch.randn(snake_case				,					generator=snake_case				,					device='''cpu'''				,					dtype=snake_case     ).to(
                        self.device     )
               else:
                    __UpperCAmelCase      :					Tuple    				=     torch.randn(snake_case				,					generator=snake_case				,					device=self.device				,					dtype=snake_case     )
          else:
               if latents.shape != latents_shape:
                    raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'     )
               __UpperCAmelCase      :					Any    				=     latents.to(self.device     )

          # set timesteps
          __UpperCAmelCase      :					List[str]    				=     '''offset''' in set(inspect.signature(self.scheduler.set_timesteps     ).parameters.keys()     )
          __UpperCAmelCase      :					List[Any]    				=     {}
          if accepts_offset:
               __UpperCAmelCase      :					Optional[int]    				=     1

          self.scheduler.set_timesteps(snake_case				,					**snake_case     )

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

          # scale the initial noise by the standard deviation required by the scheduler
          __UpperCAmelCase      :					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]
          __UpperCAmelCase      :					List[Any]    				=     '''eta''' in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
          __UpperCAmelCase      :					Any    				=     {}
          if accepts_eta:
               __UpperCAmelCase      :					Dict    				=     eta

          # check if the scheduler accepts generator
          __UpperCAmelCase      :					Union[str, Any]    				=     '''generator''' in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
          if accepts_generator:
               __UpperCAmelCase      :					Optional[Any]    				=     generator

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

               # predict the noise residual
               __UpperCAmelCase      :					List[Any]    				=     self.unet(snake_case				,					snake_case				,					encoder_hidden_states=snake_case     ).sample

               # perform classifier free guidance
               if do_classifier_free_guidance:
                    __UpperCAmelCase				,  __UpperCAmelCase      :					Optional[Any]    				=     noise_pred.chunk(2     )
                    __UpperCAmelCase      :					Tuple    				=     noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

               # perform clip guidance
               if clip_guidance_scale > 0:
                    __UpperCAmelCase      :					Optional[Any]    				=     (
                        text_embeddings.chunk(2     )[1] if do_classifier_free_guidance else text_embeddings
                    )
                    __UpperCAmelCase				,  __UpperCAmelCase      :					List[Any]    				=     self.cond_fn(
                        snake_case				,					snake_case				,					snake_case				,					snake_case				,					snake_case				,					snake_case				,					snake_case				,					snake_case				,					snake_case				,					)

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

          # scale and decode the image latents with vae
          __UpperCAmelCase      :					Any    				=     1 / self.vae.config.scaling_factor * latents
          __UpperCAmelCase      :					int    				=     self.vae.decode(snake_case     ).sample

          __UpperCAmelCase      :					int    				=     (image / 2 + 0.5).clamp(0				,					1     )
          __UpperCAmelCase      :					Union[str, Any]    				=     image.cpu().permute(0				,					2				,					3				,					1     ).numpy()

          if output_type == "pil":
               __UpperCAmelCase      :					Any    				=     self.numpy_to_pil(snake_case     )

          if not return_dict:
               return (image, None)

          return StableDiffusionPipelineOutput(images=snake_case				,					nsfw_content_detected=snake_case     )



'''simple docstring'''



import re




def       _a      (		_lowercase   :      List[Any]    ):





     '''simple docstring'''






     def docstring_decorator(_lowercase   :      str    ):
          __UpperCAmelCase      :					Tuple    				=     fn.__doc__
          __UpperCAmelCase      :					Tuple    				=     func_doc.split('''\n'''    )
          __UpperCAmelCase      :					Optional[Any]    				=     0
          while i < len(_lowercase    ) and re.search(R'''^\s*Examples?:\s*$'''     ,   lines[i]    ) is None:
               i += 1
          if i < len(_lowercase    ):
               __UpperCAmelCase      :					Optional[int]    				=     example_docstring
               __UpperCAmelCase      :					Tuple    				=     '''\n'''.join(_lowercase    )
          else:
               raise ValueError(
                   F'The function {fn} should have an empty \'Examples:\' in its docstring as placeholder, '
                   F'current docstring is:\n{func_doc}'    )
          __UpperCAmelCase      :					Tuple    				=     func_doc
          return fn

     return docstring_decorator