'''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    UpperCAmelCase__	(							nn.Module      ):






   """simple docstring"""
   def __init__(self							,					_a							,					_a=1.0					)     ->  Tuple:
        super().__init__()

        lowercase_       :							Union[str, Any]  					=       cut_size
        lowercase_       :							List[Any]  					=       cut_power





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



def _UpperCamelCase (							SCREAMING_SNAKE_CASE_	,							SCREAMING_SNAKE_CASE_				):
     lowercase_       :							Any  					=       F.normalize(SCREAMING_SNAKE_CASE_	,							dim=-1				)
     lowercase_       :							Any  					=       F.normalize(SCREAMING_SNAKE_CASE_	,							dim=-1				)
     return (x - y).norm(dim=-1				).div(2				).arcsin().pow(2				).mul(2				)

def _UpperCamelCase (							SCREAMING_SNAKE_CASE_	,							SCREAMING_SNAKE_CASE_				):
     for param in model.parameters():
          lowercase_       :							List[str]  					=       value






class    UpperCAmelCase__	(							_snake_case      ):






   """simple docstring"""
   def __init__(self							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					)     ->  Dict:
        super().__init__()
        self.register_modules(
            vae=_a							,					text_encoder=_a							,					clip_model=_a							,					tokenizer=_a							,					unet=_a							,					scheduler=_a							,					feature_extractor=_a							,					)

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

        set_requires_grad(self.text_encoder							,					_a					)
        set_requires_grad(self.clip_model							,					_a					)
   def 		_lowerCamelCase			(self							,					_a = "auto"					)     ->  Dict:
        if slice_size == "auto":
             # half the attention head size is usually a good trade-off between
             # speed and memory
             lowercase_       :							Union[str, Any]  					=       self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(_a					)
   def 		_lowerCamelCase			(self					)     ->  Optional[Any]:
        self.enable_attention_slicing(_a					)
   def 		_lowerCamelCase			(self					)     ->  Any:
        set_requires_grad(self.vae							,					_a					)
   def 		_lowerCamelCase			(self					)     ->  List[str]:
        set_requires_grad(self.vae							,					_a					)
   def 		_lowerCamelCase			(self					)     ->  List[str]:
        set_requires_grad(self.unet							,					_a					)
   def 		_lowerCamelCase			(self					)     ->  List[str]:
        set_requires_grad(self.unet							,					_a					)
   @torch.enable_grad()
   def 		_lowerCamelCase			(self							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a=True							,					)     ->  Optional[int]:
        lowercase_       :							Optional[Any]  					=       latents.detach().requires_grad_()

        lowercase_       :							Union[str, Any]  					=       self.scheduler.scale_model_input(_a							,					_a					)

        # predict the noise residual
        lowercase_       :							Any  					=       self.unet(_a							,					_a							,					encoder_hidden_states=_a					).sample

        if isinstance(self.scheduler							,					(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)					):
             lowercase_       :							Optional[int]  					=       self.scheduler.alphas_cumprod[timestep]
             lowercase_       :							Union[str, Any]  					=       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
             lowercase_       :							str  					=       (latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

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

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

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

        lowercase_       :							Tuple  					=       self.clip_model.get_image_features(_a					)
        lowercase_       :							Dict  					=       image_embeddings_clip / image_embeddings_clip.norm(p=2							,					dim=-1							,					keepdim=_a					)

        if use_cutouts:
             lowercase_       :							Tuple  					=       spherical_dist_loss(_a							,					_a					)
             lowercase_       :							str  					=       dists.view([num_cutouts, sample.shape[0], -1]					)
             lowercase_       :							Optional[int]  					=       dists.sum(2					).mean(0					).sum() * clip_guidance_scale
        else:
             lowercase_       :							List[str]  					=       spherical_dist_loss(_a							,					_a					).mean() * clip_guidance_scale

        lowercase_       :							List[str]  					=       -torch.autograd.grad(_a							,					_a					)[0]

        if isinstance(self.scheduler							,					_a					):
             lowercase_       :							Dict  					=       latents.detach() + grads * (sigma**2)
             lowercase_       :							Optional[Any]  					=       noise_pred_original
        else:
             lowercase_       :							Tuple  					=       noise_pred_original - torch.sqrt(_a					) * grads
        return noise_pred, latents





   @torch.no_grad()
   def __call__(self							,					_a							,					_a = 512							,					_a = 512							,					_a = 50							,					_a = 7.5							,					_a = 1							,					_a = 0.0							,					_a = 100							,					_a = None							,					_a = 4							,					_a = True							,					_a = None							,					_a = None							,					_a = "pil"							,					_a = True							,					)     ->  Optional[Any]:
        if isinstance(_a							,					_a					):
             lowercase_       :							Tuple  					=       1
        elif isinstance(_a							,					_a					):
             lowercase_       :							List[str]  					=       len(_a					)
        else:
             raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(_a					)}'''					)

        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
        lowercase_       :							Tuple  					=       self.tokenizer(
            _a							,					padding='max_length'							,					max_length=self.tokenizer.model_max_length							,					truncation=_a							,					return_tensors='pt'							,					)
        lowercase_       :							Dict  					=       self.text_encoder(text_input.input_ids.to(self.device					)					)[0]
        # duplicate text embeddings for each generation per prompt
        lowercase_       :							Optional[Any]  					=       text_embeddings.repeat_interleave(_a							,					dim=0					)

        if clip_guidance_scale > 0:
             if clip_prompt is not None:
                  lowercase_       :							int  					=       self.tokenizer(
                      _a							,					padding='max_length'							,					max_length=self.tokenizer.model_max_length							,					truncation=_a							,					return_tensors='pt'							,					).input_ids.to(self.device					)
             else:
                  lowercase_       :							str  					=       text_input.input_ids.to(self.device					)
             lowercase_       :							Optional[Any]  					=       self.clip_model.get_text_features(_a					)
             lowercase_       :							Union[str, Any]  					=       text_embeddings_clip / text_embeddings_clip.norm(p=2							,					dim=-1							,					keepdim=_a					)
             # duplicate text embeddings clip for each generation per prompt
             lowercase_       :							Dict  					=       text_embeddings_clip.repeat_interleave(_a							,					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.
        lowercase_       :							Optional[int]  					=       guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
             lowercase_       :							Tuple  					=       text_input.input_ids.shape[-1]
             lowercase_       :							str  					=       self.tokenizer(['']							,					padding='max_length'							,					max_length=_a							,					return_tensors='pt'					)
             lowercase_       :							List[str]  					=       self.text_encoder(uncond_input.input_ids.to(self.device					)					)[0]
             # duplicate unconditional embeddings for each generation per prompt
             lowercase_       :							Any  					=       uncond_embeddings.repeat_interleave(_a							,					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
             lowercase_       :							Optional[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`.
        lowercase_       :							Dict  					=       (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        lowercase_       :							Any  					=       text_embeddings.dtype
        if latents is None:
             if self.device.type == "mps":
                  # randn does not work reproducibly on mps
                  lowercase_       :							Any  					=       torch.randn(_a							,					generator=_a							,					device='cpu'							,					dtype=_a					).to(
                      self.device					)
             else:
                  lowercase_       :							str  					=       torch.randn(_a							,					generator=_a							,					device=self.device							,					dtype=_a					)
        else:
             if latents.shape != latents_shape:
                  raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''					)
             lowercase_       :							int  					=       latents.to(self.device					)

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

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

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

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

        # check if the scheduler accepts generator
        lowercase_       :							Tuple  					=       'generator' in set(inspect.signature(self.scheduler.step					).parameters.keys()					)
        if accepts_generator:
             lowercase_       :							str  					=       generator

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

             # predict the noise residual
             lowercase_       :							Union[str, Any]  					=       self.unet(_a							,					_a							,					encoder_hidden_states=_a					).sample

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

             # perform clip guidance
             if clip_guidance_scale > 0:
                  lowercase_       :							Optional[int]  					=       (
                      text_embeddings.chunk(2					)[1] if do_classifier_free_guidance else text_embeddings
                  )
                  lowercase_					,lowercase_       :							int  					=       self.cond_fn(
                      _a							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					_a							,					)

             # compute the previous noisy sample x_t -> x_t-1
             lowercase_       :							Dict  					=       self.scheduler.step(_a							,					_a							,					_a							,					**_a					).prev_sample

        # scale and decode the image latents with vae
        lowercase_       :							Tuple  					=       1 / self.vae.config.scaling_factor * latents
        lowercase_       :							str  					=       self.vae.decode(_a					).sample

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

        if output_type == "pil":
             lowercase_       :							str  					=       self.numpy_to_pil(_a					)

        if not return_dict:
             return (image, None)

        return StableDiffusionPipelineOutput(images=_a							,					nsfw_content_detected=_a					)





'''simple docstring'''




from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL
from PIL import Image

from ...utils import BaseOutput, is_torch_available, is_transformers_available


@dataclass
# Copied from diffusers.pipelines.stable_diffusion.__init__.StableDiffusionPipelineOutput with Stable->Alt
class    UpperCAmelCase__	(							_snake_case      ):






   """simple docstring"""
   A				:							Union[List[PIL.Image.Image], np.ndarray]
   A				:							Optional[List[bool]]


if is_transformers_available() and is_torch_available():
 from .modeling_roberta_series import RobertaSeriesModelWithTransformation
 from .pipeline_alt_diffusion import AltDiffusionPipeline
 from .pipeline_alt_diffusion_imgaimg import AltDiffusionImgaImgPipeline




