






import numpy as np




def 						__SCREAMING_SNAKE_CASE   (					_UpperCamelCase  ,						_UpperCamelCase  ,						_UpperCamelCase							):


   """simple docstring"""




   lowercase_		:      int          =       n_draws * class_counts / class_counts.sum()
   # floored means we don't overshoot n_samples, but probably undershoot
   lowercase_		:      Optional[int]          =       np.floor(_UpperCamelCase							)
   # we add samples according to how much "left over" probability
   # they had, until we arrive at n_samples
   lowercase_		:      List[str]          =       int(n_draws - floored.sum()							)
   if need_to_add > 0:
      lowercase_		:      Any          =       continuous - floored
      lowercase_		:      Optional[Any]          =       np.sort(np.unique(_UpperCamelCase							)							)[::-1]
      # add according to remainder, but break ties
      # randomly to avoid biases
      for value in values:
         (lowercase_						)		:      Any          =       np.where(remainder == value							)
         # if we need_to_add less than what's in inds
         # we draw randomly from them.
         # if we need to add more, we add them all and
         # go to the next value
         lowercase_		:      Dict          =       min(len(_UpperCamelCase							)  ,						_UpperCamelCase							)
         lowercase_		:      Union[str, Any]          =       rng.choice(_UpperCamelCase  ,						size=_UpperCamelCase  ,						replace=_UpperCamelCase							)
         floored[inds] += 1
         need_to_add -= add_now
         if need_to_add == 0:
            break
   return floored.astype(np.intaa							)





def 						__SCREAMING_SNAKE_CASE   (					_UpperCamelCase  ,						_UpperCamelCase  ,						_UpperCamelCase  ,						_UpperCamelCase  ,						_UpperCamelCase=10							):


   """simple docstring"""




   lowercase_		:      Optional[int]          =       np.unique(_UpperCamelCase  ,						return_inverse=_UpperCamelCase							)
   lowercase_		:      Optional[int]          =       classes.shape[0]
   lowercase_		:      Optional[Any]          =       np.bincount(_UpperCamelCase							)
   if np.min(_UpperCamelCase							) < 2:
      raise ValueError("Minimum class count error"							)
   if n_train < n_classes:
      raise ValueError(
          "The train_size = %d should be greater or " "equal to the number of classes = %d" % (n_train, n_classes)							)
   if n_test < n_classes:
      raise ValueError(
          "The test_size = %d should be greater or " "equal to the number of classes = %d" % (n_test, n_classes)							)
   lowercase_		:      List[Any]          =       np.split(np.argsort(_UpperCamelCase  ,						kind="mergesort"							)  ,						np.cumsum(_UpperCamelCase							)[:-1]							)
   for _ in range(_UpperCamelCase							):
      lowercase_		:      Dict          =       approximate_mode(_UpperCamelCase  ,						_UpperCamelCase  ,						_UpperCamelCase							)
      lowercase_		:      Tuple          =       class_counts - n_i
      lowercase_		:      List[str]          =       approximate_mode(_UpperCamelCase  ,						_UpperCamelCase  ,						_UpperCamelCase							)

      lowercase_		:      Any          =       []
      lowercase_		:      Union[str, Any]          =       []

      for i in range(_UpperCamelCase							):
         lowercase_		:      str          =       rng.permutation(class_counts[i]							)
         lowercase_		:      int          =       class_indices[i].take(_UpperCamelCase  ,						mode="clip"							)
         train.extend(perm_indices_class_i[: n_i[i]]							)
         test.extend(perm_indices_class_i[n_i[i] : n_i[i] + t_i[i]]							)
      lowercase_		:      Union[str, Any]          =       rng.permutation(_UpperCamelCase							)
      lowercase_		:      Union[str, Any]          =       rng.permutation(_UpperCamelCase							)

      yield train, test












'''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     ):




    def __init__(     self					:    str					,       a					:    Any					,       a					:    Optional[Any]=1.0     ):






       '''simple docstring'''






       super().__init__()

       lowercase_		:      int          =       cut_size
       lowercase_		:      Optional[Any]          =       cut_power





    def 			lowerCAmelCase__			(     self					:    Union[str, Any]					,       a					:    Optional[Any]					,       a					:    Optional[int]     ):






       '''simple docstring'''






       lowercase_      ,						lowercase_		:      int          =       pixel_values.shape[2:4]
       lowercase_		:      Union[str, Any]          =       min(a					,       a     )
       lowercase_		:      str          =       min(a					,       a					,       self.cut_size     )
       lowercase_		:      Dict          =       []
       for _ in range(a     ):
          lowercase_		:      Optional[int]          =       int(torch.rand([]     ) ** self.cut_power * (max_size - min_size) + min_size     )
          lowercase_		:      Union[str, Any]          =       torch.randint(0					,       sideX - size + 1					,       ()     )
          lowercase_		:      List[str]          =       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 						__SCREAMING_SNAKE_CASE   (					_UpperCamelCase  ,						_UpperCamelCase							):


   """simple docstring"""




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




def 						__SCREAMING_SNAKE_CASE   (					_UpperCamelCase  ,						_UpperCamelCase							):


   """simple docstring"""




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





class     _UpperCAmelCase  (						snake_case     ):




    def __init__(     self					:    int					,       a					:    AutoencoderKL					,       a					:    CLIPTextModel					,       a					:    CLIPModel					,       a					:    CLIPTokenizer					,       a					:    UNetaDConditionModel					,       a					:    Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]					,       a					:    CLIPImageProcessor					,       ):






       '''simple docstring'''






       super().__init__()
       self.register_modules(
           vae=a					,       text_encoder=a					,       clip_model=a					,       tokenizer=a					,       unet=a					,       scheduler=a					,       feature_extractor=a					,       )

       lowercase_		:      Union[str, Any]          =       transforms.Normalize(mean=feature_extractor.image_mean					,       std=feature_extractor.image_std     )
       lowercase_		:      Optional[int]          =       (
           feature_extractor.size
           if isinstance(feature_extractor.size					,       a     )
           else feature_extractor.size["shortest_edge"]
       )
       lowercase_		:      Union[str, Any]          =       MakeCutouts(self.cut_out_size     )

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




    def 			lowerCAmelCase__			(     self					:    Union[str, Any]					,       a					:    Optional[Union[str, int]] = "auto"     ):






       '''simple docstring'''






       if slice_size == "auto":
          # half the attention head size is usually a good trade-off between
          # speed and memory
          lowercase_		:      Tuple          =       self.unet.config.attention_head_dim // 2
       self.unet.set_attention_slice(a     )




    def 			lowerCAmelCase__			(     self					:    int     ):






       '''simple docstring'''






       self.enable_attention_slicing(a     )




    def 			lowerCAmelCase__			(     self					:    List[str]     ):






       '''simple docstring'''






       set_requires_grad(self.vae					,       a     )




    def 			lowerCAmelCase__			(     self					:    Optional[Any]     ):






       '''simple docstring'''






       set_requires_grad(self.vae					,       a     )




    def 			lowerCAmelCase__			(     self					:    Any     ):






       '''simple docstring'''






       set_requires_grad(self.unet					,       a     )




    def 			lowerCAmelCase__			(     self					:    Dict     ):






       '''simple docstring'''






       set_requires_grad(self.unet					,       a     )




    @torch.enable_grad()
    def 			lowerCAmelCase__			(     self					:    int					,       a					:    int					,       a					:    str					,       a					:    Dict					,       a					:    str					,       a					:    Dict					,       a					:    Dict					,       a					:    Tuple					,       a					:    Dict					,       a					:    str=True					,       ):






       '''simple docstring'''






       lowercase_		:      Dict          =       latents.detach().requires_grad_()

       lowercase_		:      List[Any]          =       self.scheduler.scale_model_input(a					,       a     )

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

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

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

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

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

       lowercase_		:      int          =       self.clip_model.get_image_features(a     )
       lowercase_		:      Any          =       image_embeddings_clip / image_embeddings_clip.norm(p=2					,       dim=-1					,       keepdim=a     )

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

       lowercase_		:      Tuple          =       -torch.autograd.grad(a					,       a     )[0]

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





    @torch.no_grad()
    def __call__(     self					:    Tuple					,       a					:    Union[str, List[str]]					,       a					:    Optional[int] = 5_1_2					,       a					:    Optional[int] = 5_1_2					,       a					:    Optional[int] = 5_0					,       a					:    Optional[float] = 7.5					,       a					:    Optional[int] = 1					,       a					:    float = 0.0					,       a					:    Optional[float] = 1_0_0					,       a					:    Optional[Union[str, List[str]]] = None					,       a					:    Optional[int] = 4					,       a					:    Optional[bool] = True					,       a					:    Optional[torch.Generator] = None					,       a					:    Optional[torch.FloatTensor] = None					,       a					:    Optional[str] = "pil"					,       a					:    bool = True					,       ):






       '''simple docstring'''






       if isinstance(a					,       a     ):
          lowercase_		:      str          =       1
       elif isinstance(a					,       a     ):
          lowercase_		:      Any          =       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_		:      str          =       self.tokenizer(
           a					,       padding="max_length"					,       max_length=self.tokenizer.model_max_length					,       truncation=a					,       return_tensors="pt"					,       )
       lowercase_		:      List[Any]          =       self.text_encoder(text_input.input_ids.to(self.device     )     )[0]
       # duplicate text embeddings for each generation per prompt
       lowercase_		:      Union[str, Any]          =       text_embeddings.repeat_interleave(a					,       dim=0     )

       if clip_guidance_scale > 0:
          if clip_prompt is not None:
             lowercase_		:      List[Any]          =       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_		:      Optional[Any]          =       text_input.input_ids.to(self.device     )
          lowercase_		:      Union[str, Any]          =       self.clip_model.get_text_features(a     )
          lowercase_		:      Dict          =       text_embeddings_clip / text_embeddings_clip.norm(p=2					,       dim=-1					,       keepdim=a     )
          # duplicate text embeddings clip for each generation per prompt
          lowercase_		:      int          =       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_		:      Union[str, Any]          =       guidance_scale > 1.0
       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance:
          lowercase_		:      List[Any]          =       text_input.input_ids.shape[-1]
          lowercase_		:      str          =       self.tokenizer([""]					,       padding="max_length"					,       max_length=a					,       return_tensors="pt"     )
          lowercase_		:      str          =       self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]
          # duplicate unconditional embeddings for each generation per prompt
          lowercase_		:      Dict          =       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_		:      Dict          =       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_		:      Optional[Any]          =       (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
       lowercase_		:      str          =       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_		:      int          =       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_		:      List[str]          =       latents.to(self.device     )

       # set timesteps
       lowercase_		:      str          =       "offset" in set(inspect.signature(self.scheduler.set_timesteps     ).parameters.keys()     )
       lowercase_		:      List[Any]          =       {}
       if accepts_offset:
          lowercase_		:      Any          =       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_		:      Union[str, Any]          =       self.scheduler.timesteps.to(self.device     )

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

       # check if the scheduler accepts generator
       lowercase_		:      str          =       "generator" in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
       if accepts_generator:
          lowercase_		:      List[Any]          =       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_		:      Optional[Any]          =       self.scheduler.scale_model_input(a					,       a     )

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

          # perform classifier free guidance
          if do_classifier_free_guidance:
             lowercase_      ,						lowercase_		:      List[Any]          =       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_		:      List[str]          =       (
                 text_embeddings.chunk(2     )[1] if do_classifier_free_guidance else text_embeddings
             )
             lowercase_      ,						lowercase_		:      List[Any]          =       self.cond_fn(
                 a					,       a					,       a					,       a					,       a					,       a					,       a					,       a					,       a					,       )

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

       # scale and decode the image latents with vae
       lowercase_		:      int          =       1 / self.vae.config.scaling_factor * latents
       lowercase_		:      List[str]          =       self.vae.decode(a     ).sample

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

       if output_type == "pil":
          lowercase_		:      List[Any]          =       self.numpy_to_pil(a     )

       if not return_dict:
          return (image, None)

       return StableDiffusionPipelineOutput(images=a					,       nsfw_content_detected=a     )




