





import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

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



def A_     (		A__		,				A__		,				A__    )  ->		Union[str, Any]:
       if isinstance(A__		,				torch.Tensor    ):
              return image
       elif isinstance(A__		,				PIL.Image.Image    ):
              a__						:     List[Any]			=    [image]

       if isinstance(image[0]		,				PIL.Image.Image    ):
              a__						:     Optional[Any]			=    [np.array(i.resize((w, h)		,				resample=PIL_INTERPOLATION['lanczos']    )    )[None, :] for i in image]
              a__						:     Optional[Any]			=    np.concatenate(A__		,				axis=0    )
              a__						:     Dict			=    np.array(A__    ).astype(np.floataa    ) / 2_55.0
              a__						:     Optional[int]			=    image.transpose(0		,				3		,				1		,				2    )
              a__						:     Union[str, Any]			=    2.0 * image - 1.0
              a__						:     Any			=    torch.from_numpy(A__    )
       elif isinstance(image[0]		,				torch.Tensor    ):
              a__						:     str			=    torch.cat(A__		,				dim=0    )
       return image



def A_     (		A__		,				A__		,				A__		,				A__=0.99_95    )  ->		Any:
       if not isinstance(A__		,				np.ndarray    ):
              a__						:     str			=    True
              a__						:     Any			=    va.device
              a__						:     Tuple			=    va.cpu().numpy()
              a__						:     Any			=    va.cpu().numpy()

       a__						:     Optional[int]			=    np.sum(va * va / (np.linalg.norm(A__    ) * np.linalg.norm(A__    ))    )
       if np.abs(A__    ) > DOT_THRESHOLD:
              a__						:     List[str]			=    (1 - t) * va + t * va
       else:
              a__						:     Any			=    np.arccos(A__    )
              a__						:     Tuple			=    np.sin(A__    )
              a__						:     Dict			=    theta_a * t
              a__						:     List[Any]			=    np.sin(A__    )
              a__						:     Union[str, Any]			=    np.sin(theta_a - theta_t    ) / sin_theta_a
              a__						:     Dict			=    sin_theta_t / sin_theta_a
              a__						:     Optional[Any]			=    sa * va + sa * va

       if inputs_are_torch:
              a__						:     Optional[int]			=    torch.from_numpy(A__    ).to(A__    )

       return va



def A_     (		A__		,				A__    )  ->		List[str]:
       a__						:     List[Any]			=    F.normalize(A__		,				dim=-1    )
       a__						:     Any			=    F.normalize(A__		,				dim=-1    )
       return (x - y).norm(dim=-1    ).div(2    ).arcsin().pow(2    ).mul(2    )



def A_     (		A__		,				A__    )  ->		Any:
       for param in model.parameters():
              a__						:     List[str]			=    value





class 			A__   (							__UpperCAmelCase     ):




   """simple docstring"""



   def __init__( self ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase=None ,  lowercase=None ,  lowercase=None ,  )      ->						Tuple:







          '''simple docstring'''

          super().__init__()
          self.register_modules(
              vae=lowercase ,  text_encoder=lowercase ,  clip_model=lowercase ,  tokenizer=lowercase ,  unet=lowercase ,  scheduler=lowercase ,  feature_extractor=lowercase ,  coca_model=lowercase ,  coca_tokenizer=lowercase ,  coca_transform=lowercase ,  )
          a__						:     Any			=    (
              feature_extractor.size
              if isinstance(feature_extractor.size ,  lowercase)
              else feature_extractor.size['shortest_edge']
          )
          a__						:     Any			=    transforms.Normalize(mean=feature_extractor.image_mean ,  std=feature_extractor.image_std)
          set_requires_grad(self.text_encoder ,  lowercase)
          set_requires_grad(self.clip_model ,  lowercase)

   def 							__lowercase		( self ,  lowercase = "auto")      ->						Optional[int]:







          '''simple docstring'''

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

   def 							__lowercase		( self)      ->						str:







          '''simple docstring'''

          self.enable_attention_slicing(lowercase)

   def 							__lowercase		( self)      ->						List[str]:







          '''simple docstring'''

          set_requires_grad(self.vae ,  lowercase)

   def 							__lowercase		( self)      ->						Tuple:







          '''simple docstring'''

          set_requires_grad(self.vae ,  lowercase)

   def 							__lowercase		( self)      ->						Optional[int]:







          '''simple docstring'''

          set_requires_grad(self.unet ,  lowercase)

   def 							__lowercase		( self)      ->						str:







          '''simple docstring'''

          set_requires_grad(self.unet ,  lowercase)

   def 							__lowercase		( self ,  lowercase ,  lowercase ,  lowercase)      ->						List[Any]:







          '''simple docstring'''

          a__						:     List[str]			=    min(int(num_inference_steps * strength) ,  lowercase)

          a__						:     Union[str, Any]			=    max(num_inference_steps - init_timestep ,  0)
          a__						:     Optional[int]			=    self.scheduler.timesteps[t_start:]

          return timesteps, num_inference_steps - t_start

   def 							__lowercase		( self ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase=None)      ->						Optional[Any]:







          '''simple docstring'''

          if not isinstance(lowercase ,  torch.Tensor):
                 raise ValueError(F'`image` has to be of type `torch.Tensor` but is {type(lowercase)}')

          a__						:     int			=    image.to(device=lowercase ,  dtype=lowercase)

          if isinstance(lowercase ,  lowercase):
                 a__						:     Optional[Any]			=    [
                     self.vae.encode(image[i : i + 1]).latent_dist.sample(generator[i]) for i in range(lowercase)
                 ]
                 a__						:     List[str]			=    torch.cat(lowercase ,  dim=0)
          else:
                 a__						:     List[str]			=    self.vae.encode(lowercase).latent_dist.sample(lowercase)

          # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
          a__						:     List[Any]			=    0.1_82_15 * init_latents
          a__						:     Union[str, Any]			=    init_latents.repeat_interleave(lowercase ,  dim=0)

          a__						:     List[Any]			=    randn_tensor(init_latents.shape ,  generator=lowercase ,  device=lowercase ,  dtype=lowercase)

          # get latents
          a__						:     Dict			=    self.scheduler.add_noise(lowercase ,  lowercase ,  lowercase)
          a__						:     int			=    init_latents

          return latents

   def 							__lowercase		( self ,  lowercase)      ->						int:







          '''simple docstring'''

          a__						:     List[Any]			=    self.coca_transform(lowercase).unsqueeze(0)
          with torch.no_grad(), torch.cuda.amp.autocast():
                 a__						:     Optional[int]			=    self.coca_model.generate(transformed_image.to(device=self.device ,  dtype=self.coca_model.dtype))
          a__						:     List[str]			=    self.coca_tokenizer.decode(generated[0].cpu().numpy())
          return generated.split('<end_of_text>')[0].replace('<start_of_text>' ,  '').rstrip(' .,')

   def 							__lowercase		( self ,  lowercase ,  lowercase)      ->						List[str]:







          '''simple docstring'''

          a__						:     str			=    self.feature_extractor.preprocess(lowercase)
          a__						:     str			=    torch.from_numpy(clip_image_input['pixel_values'][0]).unsqueeze(0).to(self.device).half()
          a__						:     Tuple			=    self.clip_model.get_image_features(lowercase)
          a__						:     Dict			=    image_embeddings_clip / image_embeddings_clip.norm(p=2 ,  dim=-1 ,  keepdim=lowercase)
          a__						:     Optional[int]			=    image_embeddings_clip.repeat_interleave(lowercase ,  dim=0)
          return image_embeddings_clip

   @torch.enable_grad()
   def 							__lowercase		( self ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  lowercase ,  )      ->						Optional[Any]:







          '''simple docstring'''

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

          a__						:     Union[str, Any]			=    self.scheduler.scale_model_input(lowercase ,  lowercase)

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

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

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

          # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
          a__						:     List[Any]			=    1 / 0.1_82_15 * sample
          a__						:     str			=    self.vae.decode(lowercase).sample
          a__						:     Union[str, Any]			=    (image / 2 + 0.5).clamp(0 ,  1)

          a__						:     str			=    transforms.Resize(self.feature_extractor_size)(lowercase)
          a__						:     Union[str, Any]			=    self.normalize(lowercase).to(latents.dtype)

          a__						:     Union[str, Any]			=    self.clip_model.get_image_features(lowercase)
          a__						:     List[Any]			=    image_embeddings_clip / image_embeddings_clip.norm(p=2 ,  dim=-1 ,  keepdim=lowercase)

          a__						:     List[str]			=    spherical_dist_loss(lowercase ,  lowercase).mean() * clip_guidance_scale

          a__						:     Optional[Any]			=    -torch.autograd.grad(lowercase ,  lowercase)[0]

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

   @torch.no_grad()
   def __call__( self ,  lowercase ,  lowercase ,  lowercase = None ,  lowercase = None ,  lowercase = 512 ,  lowercase = 512 ,  lowercase = 0.6 ,  lowercase = 50 ,  lowercase = 7.5 ,  lowercase = 1 ,  lowercase = 0.0 ,  lowercase = 100 ,  lowercase = None ,  lowercase = "pil" ,  lowercase = True ,  lowercase = 0.8 ,  lowercase = 0.1 ,  lowercase = 0.1 ,  )      ->						str:







          '''simple docstring'''

          if isinstance(lowercase ,  lowercase) and len(lowercase) != batch_size:
                 raise ValueError(F'You have passed {batch_size} batch_size, but only {len(lowercase)} generators.')

          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 isinstance(lowercase ,  torch.Generator) and batch_size > 1:
                 a__						:     Union[str, Any]			=    [generator] + [None] * (batch_size - 1)

          a__						:     List[Any]			=    [
              ('model', self.coca_model is None),
              ('tokenizer', self.coca_tokenizer is None),
              ('transform', self.coca_transform is None),
          ]
          a__						:     Optional[Any]			=    [x[0] for x in coca_is_none if x[1]]
          a__						:     Optional[int]			=    ', '.join(lowercase)
          # generate prompts with coca model if prompt is None
          if content_prompt is None:
                 if len(lowercase):
                        raise ValueError(
                            F'Content prompt is None and CoCa [{coca_is_none_str}] is None.'
                            F'Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.')
                 a__						:     Optional[Any]			=    self.get_image_description(lowercase)
          if style_prompt is None:
                 if len(lowercase):
                        raise ValueError(
                            F'Style prompt is None and CoCa [{coca_is_none_str}] is None.'
                            F' Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.')
                 a__						:     int			=    self.get_image_description(lowercase)

          # get prompt text embeddings for content and style
          a__						:     Dict			=    self.tokenizer(
              lowercase ,  padding='max_length' ,  max_length=self.tokenizer.model_max_length ,  truncation=lowercase ,  return_tensors='pt' ,  )
          a__						:     Dict			=    self.text_encoder(content_text_input.input_ids.to(self.device))[0]

          a__						:     int			=    self.tokenizer(
              lowercase ,  padding='max_length' ,  max_length=self.tokenizer.model_max_length ,  truncation=lowercase ,  return_tensors='pt' ,  )
          a__						:     Dict			=    self.text_encoder(style_text_input.input_ids.to(self.device))[0]

          a__						:     Tuple			=    slerp(lowercase ,  lowercase ,  lowercase)

          # duplicate text embeddings for each generation per prompt
          a__						:     str			=    text_embeddings.repeat_interleave(lowercase ,  dim=0)

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

          self.scheduler.set_timesteps(lowercase ,  **lowercase)
          # Some schedulers like PNDM have timesteps as arrays
          # It's more optimized to move all timesteps to correct device beforehand
          self.scheduler.timesteps.to(self.device)

          a__			,							a__						:     int			=    self.get_timesteps(lowercase ,  lowercase ,  self.device)
          a__						:     Union[str, Any]			=    timesteps[:1].repeat(lowercase)

          # Preprocess image
          a__						:     Union[str, Any]			=    preprocess(lowercase ,  lowercase ,  lowercase)
          a__						:     Optional[int]			=    self.prepare_latents(
              lowercase ,  lowercase ,  lowercase ,  text_embeddings.dtype ,  self.device ,  lowercase)

          a__						:     Dict			=    preprocess(lowercase ,  lowercase ,  lowercase)
          a__						:     Tuple			=    self.prepare_latents(
              lowercase ,  lowercase ,  lowercase ,  text_embeddings.dtype ,  self.device ,  lowercase)

          a__						:     List[str]			=    slerp(lowercase ,  lowercase ,  lowercase)

          if clip_guidance_scale > 0:
                 a__						:     Optional[Any]			=    self.get_clip_image_embeddings(lowercase ,  lowercase)
                 a__						:     List[str]			=    self.get_clip_image_embeddings(lowercase ,  lowercase)
                 a__						:     List[str]			=    slerp(
                     lowercase ,  lowercase ,  lowercase)

          # 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.
          a__						:     Any			=    guidance_scale > 1.0
          # get unconditional embeddings for classifier free guidance
          if do_classifier_free_guidance:
                 a__						:     int			=    content_text_input.input_ids.shape[-1]
                 a__						:     Union[str, Any]			=    self.tokenizer([''] ,  padding='max_length' ,  max_length=lowercase ,  return_tensors='pt')
                 a__						:     List[Any]			=    self.text_encoder(uncond_input.input_ids.to(self.device))[0]
                 # duplicate unconditional embeddings for each generation per prompt
                 a__						:     Tuple			=    uncond_embeddings.repeat_interleave(lowercase ,  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
                 a__						:     Union[str, 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`.
          a__						:     int			=    (batch_size, self.unet.config.in_channels, height // 8, width // 8)
          a__						:     List[str]			=    text_embeddings.dtype
          if latents is None:
                 if self.device.type == "mps":
                        # randn does not work reproducibly on mps
                        a__						:     List[str]			=    torch.randn(lowercase ,  generator=lowercase ,  device='cpu' ,  dtype=lowercase).to(
                            self.device)
                 else:
                        a__						:     Optional[int]			=    torch.randn(lowercase ,  generator=lowercase ,  device=self.device ,  dtype=lowercase)
          else:
                 if latents.shape != latents_shape:
                        raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {latents_shape}')
                 a__						:     Tuple			=    latents.to(self.device)

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

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

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

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

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

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

                        # compute the previous noisy sample x_t -> x_t-1
                        a__						:     Optional[Any]			=    self.scheduler.step(lowercase ,  lowercase ,  lowercase ,  **lowercase).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
          a__						:     List[str]			=    1 / 0.1_82_15 * latents
          a__						:     List[Any]			=    self.vae.decode(lowercase).sample

          a__						:     str			=    (image / 2 + 0.5).clamp(0 ,  1)
          a__						:     Dict			=    image.cpu().permute(0 ,  2 ,  3 ,  1).numpy()

          if output_type == "pil":
                 a__						:     Union[str, Any]			=    self.numpy_to_pil(lowercase)

          if not return_dict:
                 return (image, None)

          return StableDiffusionPipelineOutput(images=lowercase ,  nsfw_content_detected=lowercase)









def A_     (		A__		,				A__		,				A__    )  ->		float:
       if principal <= 0:
              raise Exception('Principal borrowed must be > 0'    )
       if rate_per_annum < 0:
              raise Exception('Rate of interest must be >= 0'    )
       if years_to_repay <= 0 or not isinstance(A__		,				A__    ):
              raise Exception('Years to repay must be an integer > 0'    )

       # Yearly rate is divided by 12 to get monthly rate
       a__						:     str			=    rate_per_annum / 12

       # Years to repay is multiplied by 12 to get number of payments as payment is monthly
       a__						:     List[Any]			=    years_to_repay * 12

       return (
           principal
           * rate_per_month
           * (1 + rate_per_month) ** number_of_payments
           / ((1 + rate_per_month) ** number_of_payments - 1)
       )


if __name__ == "__main__":
    import doctest

    doctest.testmod()


