


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

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



def  _a			(						__UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase=0.99_95 ):
     if not isinstance(__UpperCamelCase  ,  np.ndarray ):
          a_    :     Union[str, Any]										=    True
          a_    :     Any										=    va.device
          a_    :     Union[str, Any]										=    va.cpu().numpy()
          a_    :     List[str]										=    va.cpu().numpy()

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

     if inputs_are_torch:
          a_    :     Dict										=    torch.from_numpy(__UpperCamelCase ).to(__UpperCamelCase )

     return va



def  _a			(						__UpperCamelCase  ,  __UpperCamelCase ):
     a_    :     int										=    F.normalize(__UpperCamelCase  ,  dim=-1 )
     a_    :     Any										=    F.normalize(__UpperCamelCase  ,  dim=-1 )
     return (x - y).norm(dim=-1 ).div(2 ).arcsin().pow(2 ).mul(2 )



def  _a			(						__UpperCamelCase  ,  __UpperCamelCase ):
     for param in model.parameters():
          a_    :     Optional[Any]										=    value


class   a__	(						lowerCAmelCase_    ):

       def __init__(   self      :     Union[str, Any]      ,							lowerCamelCase_      :     AutoencoderKL      ,							lowerCamelCase_      :     CLIPTextModel      ,							lowerCamelCase_      :     CLIPModel      ,							lowerCamelCase_      :     CLIPTokenizer      ,							lowerCamelCase_      :     UNetaDConditionModel      ,							lowerCamelCase_      :     Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]      ,							lowerCamelCase_      :     CLIPFeatureExtractor      ,							lowerCamelCase_      :     int=None      ,							lowerCamelCase_      :     Dict=None      ,							lowerCamelCase_      :     List[Any]=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_    :     int										=    (
                feature_extractor.size
                if isinstance(feature_extractor.size      ,							lowerCamelCase_    )
                else feature_extractor.size["""shortest_edge"""]
            )
            a_    :     Union[str, Any]										=    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 						UpperCAmelCase(   self      :     Optional[int]      ,							lowerCamelCase_      :     Optional[Union[str, int]] = "auto"    ):
            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(lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Optional[Any]    ):
            self.enable_attention_slicing(lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Tuple    ):
            set_requires_grad(self.vae      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Union[str, Any]    ):
            set_requires_grad(self.vae      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Optional[int]    ):
            set_requires_grad(self.unet      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     List[str]    ):
            set_requires_grad(self.unet      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Dict      ,							lowerCamelCase_      :     Union[str, Any]      ,							lowerCamelCase_      :     Optional[Any]      ,							lowerCamelCase_      :     List[Any]    ):
            # get the original timestep using init_timestep
            a_    :     int										=    min(int(num_inference_steps * strength    )      ,							lowerCamelCase_    )

            a_    :     Dict										=    max(num_inference_steps - init_timestep      ,							0    )
            a_    :     Optional[Any]										=    self.scheduler.timesteps[t_start:]

            return timesteps, num_inference_steps - t_start

       def 						UpperCAmelCase(   self      :     List[str]      ,							lowerCamelCase_      :     List[Any]      ,							lowerCamelCase_      :     List[Any]      ,							lowerCamelCase_      :     Optional[int]      ,							lowerCamelCase_      :     Tuple      ,							lowerCamelCase_      :     List[str]      ,							lowerCamelCase_      :     int=None    ):
            if not isinstance(lowerCamelCase_      ,							torch.Tensor    ):
                 raise ValueError(F'''`image` has to be of type `torch.Tensor` but is {type(lowerCamelCase_    )}'''    )

            a_    :     str										=    image.to(device=lowerCamelCase_      ,							dtype=lowerCamelCase_    )

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

            # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
            a_    :     List[str]										=    0.1_8_2_1_5 * init_latents
            a_    :     Optional[int]										=    init_latents.repeat_interleave(lowerCamelCase_      ,							dim=0    )

            a_    :     Optional[int]										=    randn_tensor(init_latents.shape      ,							generator=lowerCamelCase_      ,							device=lowerCamelCase_      ,							dtype=lowerCamelCase_    )

            # get latents
            a_    :     Optional[Any]										=    self.scheduler.add_noise(lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_    )
            a_    :     List[Any]										=    init_latents

            return latents

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

       def 						UpperCAmelCase(   self      :     Optional[Any]      ,							lowerCamelCase_      :     Union[str, Any]      ,							lowerCamelCase_      :     Any    ):
            a_    :     Optional[int]										=    self.feature_extractor.preprocess(lowerCamelCase_    )
            a_    :     Any										=    torch.from_numpy(clip_image_input["""pixel_values"""][0]    ).unsqueeze(0    ).to(self.device    ).half()
            a_    :     Dict										=    self.clip_model.get_image_features(lowerCamelCase_    )
            a_    :     str										=    image_embeddings_clip / image_embeddings_clip.norm(p=2      ,							dim=-1      ,							keepdim=lowerCamelCase_    )
            a_    :     Dict										=    image_embeddings_clip.repeat_interleave(lowerCamelCase_      ,							dim=0    )
            return image_embeddings_clip

       @torch.enable_grad()
       def 						UpperCAmelCase(   self      :     Tuple      ,							lowerCamelCase_      :     str      ,							lowerCamelCase_      :     Optional[int]      ,							lowerCamelCase_      :     Union[str, Any]      ,							lowerCamelCase_      :     Any      ,							lowerCamelCase_      :     str      ,							lowerCamelCase_      :     int      ,							lowerCamelCase_      :     Tuple      ,							):
            a_    :     Union[str, Any]										=    latents.detach().requires_grad_()

            a_    :     List[Any]										=    self.scheduler.scale_model_input(lowerCamelCase_      ,							lowerCamelCase_    )

            # predict the noise residual
            a_    :     List[str]										=    self.unet(lowerCamelCase_      ,							lowerCamelCase_      ,							encoder_hidden_states=lowerCamelCase_    ).sample

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

                 a_    :     List[Any]										=    torch.sqrt(lowerCamelCase_    )
                 a_    :     List[Any]										=    pred_original_sample * (fac) + latents * (1 - fac)
            elif isinstance(self.scheduler      ,							lowerCamelCase_    ):
                 a_    :     List[Any]										=    self.scheduler.sigmas[index]
                 a_    :     List[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_    :     str										=    1 / 0.1_8_2_1_5 * sample
            a_    :     Optional[int]										=    self.vae.decode(lowerCamelCase_    ).sample
            a_    :     Dict										=    (image / 2 + 0.5).clamp(0      ,							1    )

            a_    :     Optional[int]										=    transforms.Resize(self.feature_extractor_size    )(lowerCamelCase_    )
            a_    :     List[Any]										=    self.normalize(lowerCamelCase_    ).to(latents.dtype    )

            a_    :     int										=    self.clip_model.get_image_features(lowerCamelCase_    )
            a_    :     int										=    image_embeddings_clip / image_embeddings_clip.norm(p=2      ,							dim=-1      ,							keepdim=lowerCamelCase_    )

            a_    :     Union[str, Any]										=    spherical_dist_loss(lowerCamelCase_      ,							lowerCamelCase_    ).mean() * clip_guidance_scale

            a_    :     Any										=    -torch.autograd.grad(lowerCamelCase_      ,							lowerCamelCase_    )[0]

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

       @torch.no_grad()
       def __call__(   self      :     List[Any]      ,							lowerCamelCase_      :     Union[torch.FloatTensor, PIL.Image.Image]      ,							lowerCamelCase_      :     Union[torch.FloatTensor, PIL.Image.Image]      ,							lowerCamelCase_      :     Optional[str] = None      ,							lowerCamelCase_      :     Optional[str] = None      ,							lowerCamelCase_      :     Optional[int] = 5_1_2      ,							lowerCamelCase_      :     Optional[int] = 5_1_2      ,							lowerCamelCase_      :     float = 0.6      ,							lowerCamelCase_      :     Optional[int] = 5_0      ,							lowerCamelCase_      :     Optional[float] = 7.5      ,							lowerCamelCase_      :     Optional[int] = 1      ,							lowerCamelCase_      :     float = 0.0      ,							lowerCamelCase_      :     Optional[float] = 1_0_0      ,							lowerCamelCase_      :     Optional[torch.Generator] = None      ,							lowerCamelCase_      :     Optional[str] = "pil"      ,							lowerCamelCase_      :     bool = True      ,							lowerCamelCase_      :     float = 0.8      ,							lowerCamelCase_      :     float = 0.1      ,							lowerCamelCase_      :     float = 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_    :     str										=    [generator] + [None] * (batch_size - 1)

            a_    :     List[str]										=    [
                ("""model""", self.coca_model is None),
                ("""tokenizer""", self.coca_tokenizer is None),
                ("""transform""", self.coca_transform is None),
            ]
            a_    :     Optional[int]										=    [x[0] for x in coca_is_none if x[1]]
            a_    :     int										=    """, """.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_    :     Tuple										=    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_    :     List[str]										=    self.get_image_description(lowerCamelCase_    )

            # get prompt text embeddings for content and style
            a_    :     Tuple										=    self.tokenizer(
                lowerCamelCase_      ,							padding="""max_length"""      ,							max_length=self.tokenizer.model_max_length      ,							truncation=lowerCamelCase_      ,							return_tensors="""pt"""      ,							)
            a_    :     Optional[int]										=    self.text_encoder(content_text_input.input_ids.to(self.device    )    )[0]

            a_    :     Optional[int]										=    self.tokenizer(
                lowerCamelCase_      ,							padding="""max_length"""      ,							max_length=self.tokenizer.model_max_length      ,							truncation=lowerCamelCase_      ,							return_tensors="""pt"""      ,							)
            a_    :     Tuple										=    self.text_encoder(style_text_input.input_ids.to(self.device    )    )[0]

            a_    :     Dict										=    slerp(lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_    )

            # duplicate text embeddings for each generation per prompt
            a_    :     Tuple										=    text_embeddings.repeat_interleave(lowerCamelCase_      ,							dim=0    )

            # set timesteps
            a_    :     Optional[int]										=    """offset""" in set(inspect.signature(self.scheduler.set_timesteps    ).parameters.keys()    )
            a_    :     Optional[int]										=    {}
            if accepts_offset:
                 a_    :     List[Any]										=    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_    :     List[Any]										=    self.get_timesteps(lowerCamelCase_      ,							lowerCamelCase_      ,							self.device    )
            a_    :     List[Any]										=    timesteps[:1].repeat(lowerCamelCase_    )

            # Preprocess image
            a_    :     str										=    preprocess(lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_    )
            a_    :     List[str]										=    self.prepare_latents(
                lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_      ,							text_embeddings.dtype      ,							self.device      ,							lowerCamelCase_    )

            a_    :     Optional[Any]										=    preprocess(lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_    )
            a_    :     List[str]										=    self.prepare_latents(
                lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_      ,							text_embeddings.dtype      ,							self.device      ,							lowerCamelCase_    )

            a_    :     Optional[int]										=    slerp(lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_    )

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

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

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

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

                      # predict the noise residual
                      a_    :     List[str]										=    self.unet(lowerCamelCase_      ,							lowerCamelCase_      ,							encoder_hidden_states=lowerCamelCase_    ).sample

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

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

                      # compute the previous noisy sample x_t -> x_t-1
                      a_    :     str										=    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_    :     Any										=    1 / 0.1_8_2_1_5 * latents
            a_    :     int										=    self.vae.decode(lowerCamelCase_    ).sample

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

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

            if not return_dict:
                 return (image, None)

            return StableDiffusionPipelineOutput(images=lowerCamelCase_      ,							nsfw_content_detected=lowerCamelCase_    )







import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402


class   a__	(						unittest.TestCase    ):

       def 						UpperCAmelCase(   self      :     int    ):
            a_    :     Any										=    0

       def 						UpperCAmelCase(   self      :     Dict    ):
            a_    :     List[str]										=    AutoImageProcessor.from_pretrained("""openai/clip-vit-base-patch32"""    )
            self.assertIsInstance(lowerCamelCase_      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     int    ):
            with tempfile.TemporaryDirectory() as tmpdirname:
                 a_    :     Any										=    Path(lowerCamelCase_    ) / """preprocessor_config.json"""
                 a_    :     Tuple										=    Path(lowerCamelCase_    ) / """config.json"""
                 json.dump(
                     {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}      ,							open(lowerCamelCase_      ,							"""w"""    )      ,							)
                 json.dump({"""model_type""": """clip"""}      ,							open(lowerCamelCase_      ,							"""w"""    )    )

                 a_    :     Union[str, Any]										=    AutoImageProcessor.from_pretrained(lowerCamelCase_    )
                 self.assertIsInstance(lowerCamelCase_      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     List[Any]    ):
            # Ensure we can load the image processor from the feature extractor config
            with tempfile.TemporaryDirectory() as tmpdirname:
                 a_    :     Dict										=    Path(lowerCamelCase_    ) / """preprocessor_config.json"""
                 a_    :     List[str]										=    Path(lowerCamelCase_    ) / """config.json"""
                 json.dump(
                     {"""feature_extractor_type""": """CLIPFeatureExtractor""", """processor_class""": """CLIPProcessor"""}      ,							open(lowerCamelCase_      ,							"""w"""    )      ,							)
                 json.dump({"""model_type""": """clip"""}      ,							open(lowerCamelCase_      ,							"""w"""    )    )

                 a_    :     int										=    AutoImageProcessor.from_pretrained(lowerCamelCase_    )
                 self.assertIsInstance(lowerCamelCase_      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Optional[Any]    ):
            with tempfile.TemporaryDirectory() as tmpdirname:
                 a_    :     int										=    CLIPConfig()

                 # Create a dummy config file with image_proceesor_type
                 a_    :     Tuple										=    Path(lowerCamelCase_    ) / """preprocessor_config.json"""
                 a_    :     Tuple										=    Path(lowerCamelCase_    ) / """config.json"""
                 json.dump(
                     {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}      ,							open(lowerCamelCase_      ,							"""w"""    )      ,							)
                 json.dump({"""model_type""": """clip"""}      ,							open(lowerCamelCase_      ,							"""w"""    )    )

                 # remove image_processor_type to make sure config.json alone is enough to load image processor locally
                 a_    :     Optional[int]										=    AutoImageProcessor.from_pretrained(lowerCamelCase_    ).to_dict()

                 config_dict.pop("""image_processor_type"""    )
                 a_    :     Tuple										=    CLIPImageProcessor(**lowerCamelCase_    )

                 # save in new folder
                 model_config.save_pretrained(lowerCamelCase_    )
                 config.save_pretrained(lowerCamelCase_    )

                 a_    :     Optional[int]										=    AutoImageProcessor.from_pretrained(lowerCamelCase_    )

                 # make sure private variable is not incorrectly saved
                 a_    :     str										=    json.loads(config.to_json_string()    )
                 self.assertTrue("""_processor_class""" not in dict_as_saved    )

            self.assertIsInstance(lowerCamelCase_      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Dict    ):
            with tempfile.TemporaryDirectory() as tmpdirname:
                 a_    :     Union[str, Any]										=    Path(lowerCamelCase_    ) / """preprocessor_config.json"""
                 json.dump(
                     {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}      ,							open(lowerCamelCase_      ,							"""w"""    )      ,							)

                 a_    :     Optional[int]										=    AutoImageProcessor.from_pretrained(lowerCamelCase_    )
                 self.assertIsInstance(lowerCamelCase_      ,							lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     int    ):
            with self.assertRaisesRegex(
                lowerCamelCase_      ,							"""clip-base is not a local folder and is not a valid model identifier"""    ):
                 a_    :     Any										=    AutoImageProcessor.from_pretrained("""clip-base"""    )

       def 						UpperCAmelCase(   self      :     Optional[Any]    ):
            with self.assertRaisesRegex(
                lowerCamelCase_      ,							R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""    ):
                 a_    :     Dict										=    AutoImageProcessor.from_pretrained(lowerCamelCase_      ,							revision="""aaaaaa"""    )

       def 						UpperCAmelCase(   self      :     Dict    ):
            with self.assertRaisesRegex(
                lowerCamelCase_      ,							"""hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""      ,							):
                 a_    :     Any										=    AutoImageProcessor.from_pretrained("""hf-internal-testing/config-no-model"""    )

       def 						UpperCAmelCase(   self      :     str    ):
            # If remote code is not set, we will time out when asking whether to load the model.
            with self.assertRaises(lowerCamelCase_    ):
                 a_    :     int										=    AutoImageProcessor.from_pretrained("""hf-internal-testing/test_dynamic_image_processor"""    )
            # If remote code is disabled, we can't load this config.
            with self.assertRaises(lowerCamelCase_    ):
                 a_    :     Tuple										=    AutoImageProcessor.from_pretrained(
                     """hf-internal-testing/test_dynamic_image_processor"""      ,							trust_remote_code=lowerCamelCase_    )

            a_    :     int										=    AutoImageProcessor.from_pretrained(
                """hf-internal-testing/test_dynamic_image_processor"""      ,							trust_remote_code=lowerCamelCase_    )
            self.assertEqual(image_processor.__class__.__name__      ,							"""NewImageProcessor"""    )

            # Test image processor can be reloaded.
            with tempfile.TemporaryDirectory() as tmp_dir:
                 image_processor.save_pretrained(lowerCamelCase_    )
                 a_    :     int										=    AutoImageProcessor.from_pretrained(lowerCamelCase_      ,							trust_remote_code=lowerCamelCase_    )
            self.assertEqual(reloaded_image_processor.__class__.__name__      ,							"""NewImageProcessor"""    )

       def 						UpperCAmelCase(   self      :     List[str]    ):
            try:
                 AutoConfig.register("""custom"""      ,							lowerCamelCase_    )
                 AutoImageProcessor.register(lowerCamelCase_      ,							lowerCamelCase_    )
                 # Trying to register something existing in the Transformers library will raise an error
                 with self.assertRaises(lowerCamelCase_    ):
                      AutoImageProcessor.register(lowerCamelCase_      ,							lowerCamelCase_    )

                 with tempfile.TemporaryDirectory() as tmpdirname:
                      a_    :     Dict										=    Path(lowerCamelCase_    ) / """preprocessor_config.json"""
                      a_    :     Union[str, Any]										=    Path(lowerCamelCase_    ) / """config.json"""
                      json.dump(
                          {"""feature_extractor_type""": """CLIPFeatureExtractor""", """processor_class""": """CLIPProcessor"""}      ,							open(lowerCamelCase_      ,							"""w"""    )      ,							)
                      json.dump({"""model_type""": """clip"""}      ,							open(lowerCamelCase_      ,							"""w"""    )    )

                      a_    :     str										=    CustomImageProcessor.from_pretrained(lowerCamelCase_    )

                 # Now that the config is registered, it can be used as any other config with the auto-API
                 with tempfile.TemporaryDirectory() as tmp_dir:
                      image_processor.save_pretrained(lowerCamelCase_    )
                      a_    :     Any										=    AutoImageProcessor.from_pretrained(lowerCamelCase_    )
                      self.assertIsInstance(lowerCamelCase_      ,							lowerCamelCase_    )

            finally:
                 if "custom" in CONFIG_MAPPING._extra_content:
                      del CONFIG_MAPPING._extra_content["custom"]
                 if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
                      del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]

       def 						UpperCAmelCase(   self      :     Union[str, Any]    ):


            class   a__	(						lowerCAmelCase_    ):
                   lowerCamelCase__:     List[Any]							    =     True

            try:
                 AutoConfig.register("""custom"""      ,							lowerCamelCase_    )
                 AutoImageProcessor.register(lowerCamelCase_      ,							lowerCamelCase_    )
                 # If remote code is not set, the default is to use local
                 a_    :     Dict										=    AutoImageProcessor.from_pretrained("""hf-internal-testing/test_dynamic_image_processor"""    )
                 self.assertEqual(image_processor.__class__.__name__      ,							"""NewImageProcessor"""    )
                 self.assertTrue(image_processor.is_local    )

                 # If remote code is disabled, we load the local one.
                 a_    :     Optional[Any]										=    AutoImageProcessor.from_pretrained(
                     """hf-internal-testing/test_dynamic_image_processor"""      ,							trust_remote_code=lowerCamelCase_    )
                 self.assertEqual(image_processor.__class__.__name__      ,							"""NewImageProcessor"""    )
                 self.assertTrue(image_processor.is_local    )

                 # If remote is enabled, we load from the Hub
                 a_    :     Dict										=    AutoImageProcessor.from_pretrained(
                     """hf-internal-testing/test_dynamic_image_processor"""      ,							trust_remote_code=lowerCamelCase_    )
                 self.assertEqual(image_processor.__class__.__name__      ,							"""NewImageProcessor"""    )
                 self.assertTrue(not hasattr(lowerCamelCase_      ,							"""is_local"""    )    )

            finally:
                 if "custom" in CONFIG_MAPPING._extra_content:
                      del CONFIG_MAPPING._extra_content["custom"]
                 if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
                      del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]



