"""simple docstring"""



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 	_lowerCamelCase(	a   ,    a   ,    a				):
       if isinstance(a   ,    torch.Tensor				):
              return image
       elif isinstance(a   ,    PIL.Image.Image				):
              __a       =    [image]

       if isinstance(image[0]   ,    PIL.Image.Image				):
              __a       =    [np.array(i.resize((w, h)   ,    resample=PIL_INTERPOLATION["lanczos"]				)				)[None, :] for i in image]
              __a       =    np.concatenate(a   ,    axis=0				)
              __a       =    np.array(a				).astype(np.floataa				) / 2_55.0
              __a       =    image.transpose(0   ,    3   ,    1   ,    2				)
              __a       =    2.0 * image - 1.0
              __a       =    torch.from_numpy(a				)
       elif isinstance(image[0]   ,    torch.Tensor				):
              __a       =    torch.cat(a   ,    dim=0				)
       return image



def 	_lowerCamelCase(	a   ,    a   ,    a   ,    a=0.99_95				):
       if not isinstance(a   ,    np.ndarray				):
              __a       =    True
              __a       =    va.device
              __a       =    va.cpu().numpy()
              __a       =    va.cpu().numpy()

       __a       =    np.sum(va * va / (np.linalg.norm(a				) * np.linalg.norm(a				))				)
       if np.abs(a				) > DOT_THRESHOLD:
              __a       =    (1 - t) * va + t * va
       else:
              __a       =    np.arccos(a				)
              __a       =    np.sin(a				)
              __a       =    theta_a * t
              __a       =    np.sin(a				)
              __a       =    np.sin(theta_a - theta_t				) / sin_theta_a
              __a       =    sin_theta_t / sin_theta_a
              __a       =    sa * va + sa * va

       if inputs_are_torch:
              __a       =    torch.from_numpy(a				).to(a				)

       return va



def 	_lowerCamelCase(	a   ,    a				):
       __a       =    F.normalize(a   ,    dim=-1				)
       __a       =    F.normalize(a   ,    dim=-1				)
       return (x - y).norm(dim=-1				).div(2				).arcsin().pow(2				).mul(2				)



def 	_lowerCamelCase(	a   ,    a				):
       for param in model.parameters():
              __a       =    value






class  snake_case__      (       snake_case_       ):


 def __init__( self    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase=None    ,		lowerCamelCase=None    ,		lowerCamelCase=None    ,		):
        super().__init__()
        self.register_modules(
            vae=lowerCamelCase    ,		text_encoder=lowerCamelCase    ,		clip_model=lowerCamelCase    ,		tokenizer=lowerCamelCase    ,		unet=lowerCamelCase    ,		scheduler=lowerCamelCase    ,		feature_extractor=lowerCamelCase    ,		coca_model=lowerCamelCase    ,		coca_tokenizer=lowerCamelCase    ,		coca_transform=lowerCamelCase    ,		)
        __a       =    (
            feature_extractor.size
            if isinstance(feature_extractor.size    ,		lowerCamelCase						)
            else feature_extractor.size["shortest_edge"]
        )
        __a       =    transforms.Normalize(mean=feature_extractor.image_mean    ,		std=feature_extractor.image_std						)
        set_requires_grad(self.text_encoder    ,		lowerCamelCase						)
        set_requires_grad(self.clip_model    ,		lowerCamelCase						)


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


 def 						a__   ( self						):
        self.enable_attention_slicing(lowerCamelCase						)


 def 						a__   ( self						):
        set_requires_grad(self.vae    ,		lowerCamelCase						)


 def 						a__   ( self						):
        set_requires_grad(self.vae    ,		lowerCamelCase						)


 def 						a__   ( self						):
        set_requires_grad(self.unet    ,		lowerCamelCase						)


 def 						a__   ( self						):
        set_requires_grad(self.unet    ,		lowerCamelCase						)


 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						):
        # get the original timestep using init_timestep
        __a       =    min(int(num_inference_steps * strength						)    ,		lowerCamelCase						)

        __a       =    max(num_inference_steps - init_timestep    ,		0						)
        __a       =    self.scheduler.timesteps[t_start:]

        return timesteps, num_inference_steps - t_start


 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase=None						):
        if not isinstance(lowerCamelCase    ,		torch.Tensor						):
               raise ValueError(F"`image` has to be of type `torch.Tensor` but is {type(lowerCamelCase						)}"						)

        __a       =    image.to(device=lowerCamelCase    ,		dtype=lowerCamelCase						)

        if isinstance(lowerCamelCase    ,		lowerCamelCase						):
               __a       =    [
                   self.vae.encode(image[i : i + 1]						).latent_dist.sample(generator[i]						) for i in range(lowerCamelCase						)
               ]
               __a       =    torch.cat(lowerCamelCase    ,		dim=0						)
        else:
               __a       =    self.vae.encode(lowerCamelCase						).latent_dist.sample(lowerCamelCase						)

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
        __a       =    0.1_8215 * init_latents
        __a       =    init_latents.repeat_interleave(lowerCamelCase    ,		dim=0						)

        __a       =    randn_tensor(init_latents.shape    ,		generator=lowerCamelCase    ,		device=lowerCamelCase    ,		dtype=lowerCamelCase						)

        # get latents
        __a       =    self.scheduler.add_noise(lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						)
        __a       =    init_latents

        return latents


 def 						a__   ( self    ,		lowerCamelCase						):
        __a       =    self.coca_transform(lowerCamelCase						).unsqueeze(0						)
        with torch.no_grad(), torch.cuda.amp.autocast():
               __a       =    self.coca_model.generate(transformed_image.to(device=self.device    ,		dtype=self.coca_model.dtype						)						)
        __a       =    self.coca_tokenizer.decode(generated[0].cpu().numpy()						)
        return generated.split("<end_of_text>"						)[0].replace("<start_of_text>"    ,		""						).rstrip(" .,"						)


 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase						):
        __a       =    self.feature_extractor.preprocess(lowerCamelCase						)
        __a       =    torch.from_numpy(clip_image_input["pixel_values"][0]						).unsqueeze(0						).to(self.device						).half()
        __a       =    self.clip_model.get_image_features(lowerCamelCase						)
        __a       =    image_embeddings_clip / image_embeddings_clip.norm(p=2    ,		dim=-1    ,		keepdim=lowerCamelCase						)
        __a       =    image_embeddings_clip.repeat_interleave(lowerCamelCase    ,		dim=0						)
        return image_embeddings_clip


 @torch.enable_grad()
 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		):
        __a       =    latents.detach().requires_grad_()

        __a       =    self.scheduler.scale_model_input(lowerCamelCase    ,		lowerCamelCase						)

        # predict the noise residual
        __a       =    self.unet(lowerCamelCase    ,		lowerCamelCase    ,		encoder_hidden_states=lowerCamelCase						).sample

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

               __a       =    torch.sqrt(lowerCamelCase						)
               __a       =    pred_original_sample * (fac) + latents * (1 - fac)
        elif isinstance(self.scheduler    ,		lowerCamelCase						):
               __a       =    self.scheduler.sigmas[index]
               __a       =    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       =    1 / 0.1_8215 * sample
        __a       =    self.vae.decode(lowerCamelCase						).sample
        __a       =    (image / 2 + 0.5).clamp(0    ,		1						)

        __a       =    transforms.Resize(self.feature_extractor_size						)(lowerCamelCase						)
        __a       =    self.normalize(lowerCamelCase						).to(latents.dtype						)

        __a       =    self.clip_model.get_image_features(lowerCamelCase						)
        __a       =    image_embeddings_clip / image_embeddings_clip.norm(p=2    ,		dim=-1    ,		keepdim=lowerCamelCase						)

        __a       =    spherical_dist_loss(lowerCamelCase    ,		lowerCamelCase						).mean() * clip_guidance_scale

        __a       =    -torch.autograd.grad(lowerCamelCase    ,		lowerCamelCase						)[0]

        if isinstance(self.scheduler    ,		lowerCamelCase						):
               __a       =    latents.detach() + grads * (sigma**2)
               __a       =    noise_pred_original
        else:
               __a       =    noise_pred_original - torch.sqrt(lowerCamelCase						) * grads
        return noise_pred, latents






 @torch.no_grad()
 def __call__( self    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase = None    ,		lowerCamelCase = None    ,		lowerCamelCase = 512    ,		lowerCamelCase = 512    ,		lowerCamelCase = 0.6    ,		lowerCamelCase = 50    ,		lowerCamelCase = 7.5    ,		lowerCamelCase = 1    ,		lowerCamelCase = 0.0    ,		lowerCamelCase = 100    ,		lowerCamelCase = None    ,		lowerCamelCase = "pil"    ,		lowerCamelCase = True    ,		lowerCamelCase = 0.8    ,		lowerCamelCase = 0.1    ,		lowerCamelCase = 0.1    ,		):
        if isinstance(lowerCamelCase    ,		lowerCamelCase						) and len(lowerCamelCase						) != batch_size:
               raise ValueError(F"You have passed {batch_size} batch_size, but only {len(lowerCamelCase						)} 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(lowerCamelCase    ,		torch.Generator						) and batch_size > 1:
               __a       =    [generator] + [None] * (batch_size - 1)

        __a       =    [
            ("model", self.coca_model is None),
            ("tokenizer", self.coca_tokenizer is None),
            ("transform", self.coca_transform is None),
        ]
        __a       =    [x[0] for x in coca_is_none if x[1]]
        __a       =    ", ".join(lowerCamelCase						)
        # generate prompts with coca model if prompt is None
        if content_prompt is None:
               if len(lowerCamelCase						):
                      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       =    self.get_image_description(lowerCamelCase						)
        if style_prompt is None:
               if len(lowerCamelCase						):
                      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       =    self.get_image_description(lowerCamelCase						)

        # get prompt text embeddings for content and style
        __a       =    self.tokenizer(
            lowerCamelCase    ,		padding="max_length"    ,		max_length=self.tokenizer.model_max_length    ,		truncation=lowerCamelCase    ,		return_tensors="pt"    ,		)
        __a       =    self.text_encoder(content_text_input.input_ids.to(self.device						)						)[0]

        __a       =    self.tokenizer(
            lowerCamelCase    ,		padding="max_length"    ,		max_length=self.tokenizer.model_max_length    ,		truncation=lowerCamelCase    ,		return_tensors="pt"    ,		)
        __a       =    self.text_encoder(style_text_input.input_ids.to(self.device						)						)[0]

        __a       =    slerp(lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						)

        # duplicate text embeddings for each generation per prompt
        __a       =    text_embeddings.repeat_interleave(lowerCamelCase    ,		dim=0						)

        # set timesteps
        __a       =    "offset" in set(inspect.signature(self.scheduler.set_timesteps						).parameters.keys()						)
        __a       =    {}
        if accepts_offset:
               __a       =    1

        self.scheduler.set_timesteps(lowerCamelCase    ,		**lowerCamelCase						)
        # 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       =    self.get_timesteps(lowerCamelCase    ,		lowerCamelCase    ,		self.device						)
        __a       =    timesteps[:1].repeat(lowerCamelCase						)

        # Preprocess image
        __a       =    preprocess(lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						)
        __a       =    self.prepare_latents(
            lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		text_embeddings.dtype    ,		self.device    ,		lowerCamelCase						)

        __a       =    preprocess(lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						)
        __a       =    self.prepare_latents(
            lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		text_embeddings.dtype    ,		self.device    ,		lowerCamelCase						)

        __a       =    slerp(lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						)

        if clip_guidance_scale > 0:
               __a       =    self.get_clip_image_embeddings(lowerCamelCase    ,		lowerCamelCase						)
               __a       =    self.get_clip_image_embeddings(lowerCamelCase    ,		lowerCamelCase						)
               __a       =    slerp(
                   lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						)

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

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

        # check if the scheduler accepts generator
        __a       =    "generator" in set(inspect.signature(self.scheduler.step						).parameters.keys()						)
        if accepts_generator:
               __a       =    generator

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

                      # predict the noise residual
                      __a       =    self.unet(lowerCamelCase    ,		lowerCamelCase    ,		encoder_hidden_states=lowerCamelCase						).sample

                      # perform classifier free guidance
                      if do_classifier_free_guidance:
                             __a ,   __a       =    noise_pred.chunk(2						)
                             __a       =    noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

                      # compute the previous noisy sample x_t -> x_t-1
                      __a       =    self.scheduler.step(lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		**lowerCamelCase						).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
        __a       =    1 / 0.1_8215 * latents
        __a       =    self.vae.decode(lowerCamelCase						).sample

        __a       =    (image / 2 + 0.5).clamp(0    ,		1						)
        __a       =    image.cpu().permute(0    ,		2    ,		3    ,		1						).numpy()

        if output_type == "pil":
               __a       =    self.numpy_to_pil(lowerCamelCase						)

        if not return_dict:
               return (image, None)

        return StableDiffusionPipelineOutput(images=lowerCamelCase    ,		nsfw_content_detected=lowerCamelCase						)

"""simple docstring"""



import unittest

import torch
from torch import nn

from accelerate.test_utils import require_cuda
from accelerate.utils.memory import find_executable_batch_size, release_memory



def 	_lowerCamelCase(	):
       raise RuntimeError("CUDA out of memory."				)






class  snake_case__      (       nn.Module       ):


 def __init__( self						):
        super().__init__()
        __a       =    nn.Linear(3    ,		4						)
        __a       =    nn.BatchNormad(4						)
        __a       =    nn.Linear(4    ,		5						)






 def 						a__   ( self    ,		lowerCamelCase						):
        return self.lineara(self.batchnorm(self.lineara(lowerCamelCase						)						)						)






class  snake_case__      (       unittest.TestCase       ):


 def 						a__   ( self						):
        __a       =    []

        @find_executable_batch_size(starting_batch_size=128						)
        def mock_training_loop_function(lowerCamelCase						):
               nonlocal batch_sizes
               batch_sizes.append(lowerCamelCase						)
               if batch_size != 8:
                      raise_fake_out_of_memory()

        mock_training_loop_function()
        self.assertListEqual(lowerCamelCase    ,		[128, 64, 32, 16, 8]						)


 def 						a__   ( self						):
        __a       =    []

        @find_executable_batch_size(starting_batch_size=128						)
        def mock_training_loop_function(lowerCamelCase    ,		lowerCamelCase						):
               nonlocal batch_sizes
               batch_sizes.append(lowerCamelCase						)
               if batch_size != 8:
                      raise_fake_out_of_memory()
               return batch_size, arga

        __a ,   __a       =    mock_training_loop_function("hello"						)
        self.assertListEqual(lowerCamelCase    ,		[128, 64, 32, 16, 8]						)
        self.assertListEqual([bs, arga]    ,		[8, "hello"]						)


 def 						a__   ( self						):
        @find_executable_batch_size(starting_batch_size=0						)
        def mock_training_loop_function(lowerCamelCase						):
               pass

        with self.assertRaises(lowerCamelCase						) as cm:
               mock_training_loop_function()
               self.assertIn("No executable batch size found, reached zero."    ,		cm.exception.args[0]						)


 def 						a__   ( self						):
        @find_executable_batch_size(starting_batch_size=16						)
        def mock_training_loop_function(lowerCamelCase						):
               if batch_size > 0:
                      raise_fake_out_of_memory()
               pass

        with self.assertRaises(lowerCamelCase						) as cm:
               mock_training_loop_function()
               self.assertIn("No executable batch size found, reached zero."    ,		cm.exception.args[0]						)


 def 						a__   ( self						):
        @find_executable_batch_size(starting_batch_size=128						)
        def mock_training_loop_function(lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase						):
               if batch_size != 8:
                      raise raise_fake_out_of_memory()

        with self.assertRaises(lowerCamelCase						) as cm:
               mock_training_loop_function(128    ,		"hello"    ,		"world"						)
               self.assertIn("Batch size was passed into `f`"    ,		cm.exception.args[0]						)
               self.assertIn("`f(arg1='hello', arg2='world')"    ,		cm.exception.args[0]						)


 def 						a__   ( self						):
        @find_executable_batch_size(starting_batch_size=16						)
        def mock_training_loop_function(lowerCamelCase						):
               raise ValueError("Oops, we had an error!"						)

        with self.assertRaises(lowerCamelCase						) as cm:
               mock_training_loop_function()
               self.assertIn("Oops, we had an error!"    ,		cm.exception.args[0]						)






 @require_cuda
 def 						a__   ( self						):
        __a       =    torch.cuda.memory_allocated()
        __a       =    ModelForTest()
        model.cuda()
        self.assertGreater(torch.cuda.memory_allocated()    ,		lowerCamelCase						)
        __a       =    release_memory(lowerCamelCase						)
        self.assertEqual(torch.cuda.memory_allocated()    ,		lowerCamelCase						)
