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

 def __init__(				self			:							List[str]     ,   snake_case			:							Any     ,   snake_case			:							Any=1.0    )     ->		str:

  """simple docstring"""





  super().__init__()

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


 def 		SCREAMING_SNAKE_CASE__     (				self			:							Dict     ,   snake_case			:							Optional[Any]     ,   snake_case			:							Union[str, Any]    )     ->		List[str]:

  """simple docstring"""





  UpperCamelCase_      :		Dict				=      pixel_values.shape[2:4]
  UpperCamelCase_      :		Dict				=      min(snake_case     ,   snake_case    )
  UpperCamelCase_      :		List[str]				=      min(snake_case     ,   snake_case     ,   self.cut_size    )
  UpperCamelCase_      :		int				=      []
  for _ in range(snake_case    ):
   UpperCamelCase_      :		Any				=      int(torch.rand([]    ) ** self.cut_power * (max_size - min_size) + min_size    )
   UpperCamelCase_      :		int				=      torch.randint(0     ,   sideX - size + 1     ,   ()    )
   UpperCamelCase_      :		List[str]				=      torch.randint(0     ,   sideY - size + 1     ,   ()    )
   UpperCamelCase_      :		List[str]				=      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      __lowercase  (		lowerCamelCase      :  Optional[Any]						,						lowerCamelCase      :  str			):
 UpperCamelCase_      :		Optional[Any]				=      F.normalize(lowerCamelCase						,						dim=-1			)
 UpperCamelCase_      :		Tuple				=      F.normalize(lowerCamelCase						,						dim=-1			)
 return (x - y).norm(dim=-1			).div(2			).arcsin().pow(2			).mul(2			)





def      __lowercase  (		lowerCamelCase      :  List[Any]						,						lowerCamelCase      :  Union[str, Any]			):
 for param in model.parameters():
  UpperCamelCase_      :		List[Any]				=      value






class 	_lowercase		(      snake_case_ ):

 def __init__(				self			:							Union[str, Any]     ,   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     ,   )     ->		Union[str, Any]:

  """simple docstring"""





  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_      :		Dict				=      transforms.Normalize(mean=feature_extractor.image_mean     ,   std=feature_extractor.image_std    )
  UpperCamelCase_      :		Tuple				=      (
      feature_extractor.size
      if isinstance(feature_extractor.size     ,   snake_case    )
      else feature_extractor.size['shortest_edge']
  )
  UpperCamelCase_      :		Dict				=      MakeCutouts(self.cut_out_size    )

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

 def 		SCREAMING_SNAKE_CASE__     (				self			:							str     ,   snake_case			:							Optional[Union[str, int]] = "auto"    )     ->		List[Any]:

  """simple docstring"""





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

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Dict    )     ->		str:

  """simple docstring"""





  self.enable_attention_slicing(snake_case    )

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Dict    )     ->		int:

  """simple docstring"""





  set_requires_grad(self.vae     ,   snake_case    )

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Optional[int]    )     ->		str:

  """simple docstring"""





  set_requires_grad(self.vae     ,   snake_case    )

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Union[str, Any]    )     ->		Optional[Any]:

  """simple docstring"""





  set_requires_grad(self.unet     ,   snake_case    )

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Optional[Any]    )     ->		Any:

  """simple docstring"""





  set_requires_grad(self.unet     ,   snake_case    )

 @torch.enable_grad()
 def 		SCREAMING_SNAKE_CASE__     (				self			:							str     ,   snake_case			:							Union[str, Any]     ,   snake_case			:							Any     ,   snake_case			:							List[Any]     ,   snake_case			:							str     ,   snake_case			:							Dict     ,   snake_case			:							int     ,   snake_case			:							Dict     ,   snake_case			:							Tuple     ,   snake_case			:							Tuple=True     ,   )     ->		List[Any]:

  """simple docstring"""





  UpperCamelCase_      :		List[Any]				=      latents.detach().requires_grad_()

  UpperCamelCase_      :		str				=      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_      :		Any				=      self.scheduler.alphas_cumprod[timestep]
   UpperCamelCase_      :		List[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
   UpperCamelCase_      :		List[Any]				=      (latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

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

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

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

  UpperCamelCase_      :		int				=      self.clip_model.get_image_features(snake_case    )
  UpperCamelCase_      :		Tuple				=      image_embeddings_clip / image_embeddings_clip.norm(p=2     ,   dim=-1     ,   keepdim=snake_case    )

  if use_cutouts:
   UpperCamelCase_      :		Tuple				=      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_      :		Any				=      spherical_dist_loss(snake_case     ,   snake_case    ).mean() * clip_guidance_scale

  UpperCamelCase_      :		Union[str, Any]				=      -torch.autograd.grad(snake_case     ,   snake_case    )[0]

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


 @torch.no_grad()
 def __call__(				self			:							Optional[Any]     ,   snake_case			:							Union[str, List[str]]     ,   snake_case			:							Optional[int] = 5_1_2     ,   snake_case			:							Optional[int] = 5_1_2     ,   snake_case			:							Optional[int] = 5_0     ,   snake_case			:							Optional[float] = 7.5     ,   snake_case			:							Optional[int] = 1     ,   snake_case			:							float = 0.0     ,   snake_case			:							Optional[float] = 1_0_0     ,   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     ,   )     ->		Any:

  """simple docstring"""





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

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

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

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

  # check if the scheduler accepts generator
  UpperCamelCase_      :		Dict				=      '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_      :		Union[str, Any]				=      torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents
   UpperCamelCase_      :		Optional[Any]				=      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

   # perform classifier free guidance
   if do_classifier_free_guidance:
    UpperCamelCase_      :		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_      :		int				=      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_      :		List[str]				=      self.scheduler.step(snake_case     ,   snake_case     ,   snake_case     ,   **snake_case    ).prev_sample

  # scale and decode the image latents with vae
  UpperCamelCase_      :		Dict				=      1 / self.vae.config.scaling_factor * latents
  UpperCamelCase_      :		Optional[int]				=      self.vae.decode(snake_case    ).sample

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

  if output_type == "pil":
   UpperCamelCase_      :		List[str]				=      self.numpy_to_pil(snake_case    )

  if not return_dict:
   return (image, None)

  return StableDiffusionPipelineOutput(images=snake_case     ,   nsfw_content_detected=snake_case    )




def      __lowercase  (		lowerCamelCase      :  int = 998001			):
 UpperCamelCase_      :		int				=      0
 for i in range(999						,						99						,						-1			):  # 3 digit numbers range from 999 down to 100
  for j in range(999						,						99						,						-1			):
   UpperCamelCase_      :		str				=      str(i * j			)
   if product_string == product_string[::-1] and i * j < n:
    UpperCamelCase_      :		str				=      max(lowerCamelCase						,						i * j			)
 return answer


if __name__ == "__main__":
   print(F"""{solution() = }""")



