"""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        a_  (   _lowerCAmelCase						:				Optional[int]     ,				_lowerCAmelCase						:				List[str]     ,				_lowerCAmelCase						:				List[str]  ):





       '''simple docstring'''

       if isinstance(_lowerCAmelCase     ,				torch.Tensor  ):
              return image
       elif isinstance(_lowerCAmelCase     ,				PIL.Image.Image  ):
              lowercase__    :	Union[str, Any]     					=			[image]

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





def        a_  (   _lowerCAmelCase						:				Any     ,				_lowerCAmelCase						:				Optional[int]     ,				_lowerCAmelCase						:				int     ,				_lowerCAmelCase						:				Tuple=0.9_9_9_5  ):





       '''simple docstring'''

       if not isinstance(_lowerCAmelCase     ,				np.ndarray  ):
              lowercase__    :	Optional[int]     					=			True
              lowercase__    :	str     					=			va.device
              lowercase__    :	Tuple     					=			va.cpu().numpy()
              lowercase__    :	Any     					=			va.cpu().numpy()

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

       if inputs_are_torch:
              lowercase__    :	List[str]     					=			torch.from_numpy(_lowerCAmelCase  ).to(_lowerCAmelCase  )

       return va





def        a_  (   _lowerCAmelCase						:				int     ,				_lowerCAmelCase						:				Tuple  ):





       '''simple docstring'''

       lowercase__    :	Any     					=			F.normalize(_lowerCAmelCase     ,				dim=-1  )
       lowercase__    :	Dict     					=			F.normalize(_lowerCAmelCase     ,				dim=-1  )
       return (x - y).norm(dim=-1  ).div(2  ).arcsin().pow(2  ).mul(2  )





def        a_  (   _lowerCAmelCase						:				List[str]     ,				_lowerCAmelCase						:				Union[str, Any]  ):





       '''simple docstring'''

       for param in model.parameters():
              lowercase__    :	Optional[int]     					=			value






class   UpperCAmelCase_			(					_a):


      def __init__(     self	,    a	,    a	,    a	,    a	,    a	,    a	,    a	,    a=None	,    a=None	,    a=None	,    ) ->      Any:
             super().__init__()
             self.register_modules(
                 vae=a	,    text_encoder=a	,    clip_model=a	,    tokenizer=a	,    unet=a	,    scheduler=a	,    feature_extractor=a	,    coca_model=a	,    coca_tokenizer=a	,    coca_transform=a	,    )
             lowercase__    :	Tuple     					=			(
                 feature_extractor.size
                 if isinstance(feature_extractor.size	,    a							)
                 else feature_extractor.size['shortest_edge']
             )
             lowercase__    :	Optional[int]     					=			transforms.Normalize(mean=feature_extractor.image_mean	,    std=feature_extractor.image_std							)
             set_requires_grad(self.text_encoder	,    a							)
             set_requires_grad(self.clip_model	,    a							)


      def 	_UpperCAmelCase (     self	,    a = "auto"							) ->      List[Any]:
             if slice_size == "auto":
                    # half the attention head size is usually a good trade-off between
                    # speed and memory
                    lowercase__    :	List[Any]     					=			self.unet.config.attention_head_dim // 2
             self.unet.set_attention_slice(a							)


      def 	_UpperCAmelCase (     self							) ->      Dict:
             self.enable_attention_slicing(a							)


      def 	_UpperCAmelCase (     self							) ->      Any:
             set_requires_grad(self.vae	,    a							)


      def 	_UpperCAmelCase (     self							) ->      Optional[Any]:
             set_requires_grad(self.vae	,    a							)


      def 	_UpperCAmelCase (     self							) ->      int:
             set_requires_grad(self.unet	,    a							)


      def 	_UpperCAmelCase (     self							) ->      Any:
             set_requires_grad(self.unet	,    a							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a							) ->      int:
             # get the original timestep using init_timestep
             lowercase__    :	int     					=			min(int(num_inference_steps * strength							)	,    a							)

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

             return timesteps, num_inference_steps - t_start


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    a	,    a	,    a=None							) ->      Tuple:
             if not isinstance(a	,    torch.Tensor							):
                    raise ValueError(f"""`image` has to be of type `torch.Tensor` but is {type(a							)}"""							)

             lowercase__    :	str     					=			image.to(device=a	,    dtype=a							)

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

             # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
             lowercase__    :	int     					=			0.18_215 * init_latents
             lowercase__    :	int     					=			init_latents.repeat_interleave(a	,    dim=0							)

             lowercase__    :	Dict     					=			randn_tensor(init_latents.shape	,    generator=a	,    device=a	,    dtype=a							)

             # get latents
             lowercase__    :	int     					=			self.scheduler.add_noise(a	,    a	,    a							)
             lowercase__    :	List[Any]     					=			init_latents

             return latents


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


      def 	_UpperCAmelCase (     self	,    a	,    a							) ->      List[Any]:
             lowercase__    :	int     					=			self.feature_extractor.preprocess(a							)
             lowercase__    :	Optional[Any]     					=			torch.from_numpy(clip_image_input['pixel_values'][0]							).unsqueeze(0							).to(self.device							).half()
             lowercase__    :	List[str]     					=			self.clip_model.get_image_features(a							)
             lowercase__    :	Tuple     					=			image_embeddings_clip / image_embeddings_clip.norm(p=2	,    dim=-1	,    keepdim=a							)
             lowercase__    :	Optional[Any]     					=			image_embeddings_clip.repeat_interleave(a	,    dim=0							)
             return image_embeddings_clip


      @torch.enable_grad()
      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    a	,    a	,    a	,    a	,    ) ->      str:
             lowercase__    :	Tuple     					=			latents.detach().requires_grad_()

             lowercase__    :	Dict     					=			self.scheduler.scale_model_input(a	,    a							)

             # predict the noise residual
             lowercase__    :	int     					=			self.unet(a	,    a	,    encoder_hidden_states=a							).sample

             if isinstance(self.scheduler	,    (PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)							):
                    lowercase__    :	int     					=			self.scheduler.alphas_cumprod[timestep]
                    lowercase__    :	Tuple     					=			1 - alpha_prod_t
                    # compute predicted original sample from predicted noise also called
                    # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
                    lowercase__    :	Any     					=			(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

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

             lowercase__    :	Dict     					=			transforms.Resize(self.feature_extractor_size							)(a							)
             lowercase__    :	Optional[int]     					=			self.normalize(a							).to(latents.dtype							)

             lowercase__    :	str     					=			self.clip_model.get_image_features(a							)
             lowercase__    :	Tuple     					=			image_embeddings_clip / image_embeddings_clip.norm(p=2	,    dim=-1	,    keepdim=a							)

             lowercase__    :	Dict     					=			spherical_dist_loss(a	,    a							).mean() * clip_guidance_scale

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

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


      @torch.no_grad()
      def __call__(     self	,    a	,    a	,    a = None	,    a = None	,    a = 5_1_2	,    a = 5_1_2	,    a = 0.6	,    a = 5_0	,    a = 7.5	,    a = 1	,    a = 0.0	,    a = 1_0_0	,    a = None	,    a = "pil"	,    a = True	,    a = 0.8	,    a = 0.1	,    a = 0.1	,    ) ->      Dict:
             if isinstance(a	,    a							) and len(a							) != batch_size:
                    raise ValueError(f"""You have passed {batch_size} batch_size, but only {len(a							)} 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(a	,    torch.Generator							) and batch_size > 1:
                    lowercase__    :	Optional[int]     					=			[generator] + [None] * (batch_size - 1)

             lowercase__    :	Tuple     					=			[
                 ('model', self.coca_model is None),
                 ('tokenizer', self.coca_tokenizer is None),
                 ('transform', self.coca_transform is None),
             ]
             lowercase__    :	Optional[Any]     					=			[x[0] for x in coca_is_none if x[1]]
             lowercase__    :	Dict     					=			', '.join(a							)
             # generate prompts with coca model if prompt is None
             if content_prompt is None:
                    if len(a							):
                           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."""							)
                    lowercase__    :	int     					=			self.get_image_description(a							)
             if style_prompt is None:
                    if len(a							):
                           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."""							)
                    lowercase__    :	List[str]     					=			self.get_image_description(a							)

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

             lowercase__    :	Tuple     					=			self.tokenizer(
                 a	,    padding='max_length'	,    max_length=self.tokenizer.model_max_length	,    truncation=a	,    return_tensors='pt'	,    )
             lowercase__    :	Union[str, Any]     					=			self.text_encoder(style_text_input.input_ids.to(self.device							)							)[0]

             lowercase__    :	Union[str, Any]     					=			slerp(a	,    a	,    a							)

             # duplicate text embeddings for each generation per prompt
             lowercase__    :	Any     					=			text_embeddings.repeat_interleave(a	,    dim=0							)

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

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

             lowercase__					,   lowercase__    :	Optional[int]     					=			self.get_timesteps(a	,    a	,    self.device							)
             lowercase__    :	List[Any]     					=			timesteps[:1].repeat(a							)

             # Preprocess image
             lowercase__    :	Tuple     					=			preprocess(a	,    a	,    a							)
             lowercase__    :	int     					=			self.prepare_latents(
                 a	,    a	,    a	,    text_embeddings.dtype	,    self.device	,    a							)

             lowercase__    :	List[Any]     					=			preprocess(a	,    a	,    a							)
             lowercase__    :	Union[str, Any]     					=			self.prepare_latents(
                 a	,    a	,    a	,    text_embeddings.dtype	,    self.device	,    a							)

             lowercase__    :	str     					=			slerp(a	,    a	,    a							)

             if clip_guidance_scale > 0:
                    lowercase__    :	List[str]     					=			self.get_clip_image_embeddings(a	,    a							)
                    lowercase__    :	Any     					=			self.get_clip_image_embeddings(a	,    a							)
                    lowercase__    :	Dict     					=			slerp(
                        a	,    a	,    a							)

             # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
             # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
             # corresponds to doing no classifier free guidance.
             lowercase__    :	Tuple     					=			guidance_scale > 1.0
             # get unconditional embeddings for classifier free guidance
             if do_classifier_free_guidance:
                    lowercase__    :	Optional[int]     					=			content_text_input.input_ids.shape[-1]
                    lowercase__    :	Optional[Any]     					=			self.tokenizer(['']	,    padding='max_length'	,    max_length=a	,    return_tensors='pt'							)
                    lowercase__    :	Optional[int]     					=			self.text_encoder(uncond_input.input_ids.to(self.device							)							)[0]
                    # duplicate unconditional embeddings for each generation per prompt
                    lowercase__    :	Optional[int]     					=			uncond_embeddings.repeat_interleave(a	,    dim=0							)

                    # For classifier free guidance, we need to do two forward passes.
                    # Here we concatenate the unconditional and text embeddings into a single batch
                    # to avoid doing two forward passes
                    lowercase__    :	Tuple     					=			torch.cat([uncond_embeddings, text_embeddings]							)

             # get the initial random noise unless the user supplied it

             # Unlike in other pipelines, latents need to be generated in the target device
             # for 1-to-1 results reproducibility with the CompVis implementation.
             # However this currently doesn't work in `mps`.
             lowercase__    :	Tuple     					=			(batch_size, self.unet.config.in_channels, height // 8, width // 8)
             lowercase__    :	str     					=			text_embeddings.dtype
             if latents is None:
                    if self.device.type == "mps":
                           # randn does not work reproducibly on mps
                           lowercase__    :	Tuple     					=			torch.randn(a	,    generator=a	,    device='cpu'	,    dtype=a							).to(
                               self.device							)
                    else:
                           lowercase__    :	Any     					=			torch.randn(a	,    generator=a	,    device=self.device	,    dtype=a							)
             else:
                    if latents.shape != latents_shape:
                           raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""							)
                    lowercase__    :	Tuple     					=			latents.to(self.device							)

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

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

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

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

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

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

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

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
             lowercase__    :	int     					=			1 / 0.18_215 * latents
             lowercase__    :	List[Any]     					=			self.vae.decode(a							).sample

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

             if output_type == "pil":
                    lowercase__    :	Tuple     					=			self.numpy_to_pil(a							)

             if not return_dict:
                    return (image, None)

             return StableDiffusionPipelineOutput(images=a	,    nsfw_content_detected=a							)


"""simple docstring"""





def        a_  (   _lowerCAmelCase						:				str  ):





       '''simple docstring'''

       lowercase__    :	Any     					=			[0] * len(_lowerCAmelCase  )

       for i in range(1     ,				len(_lowerCAmelCase  )  ):
              # use last results for better performance - dynamic programming
              lowercase__    :	List[str]     					=			prefix_result[i - 1]
              while j > 0 and input_string[i] != input_string[j]:
                     lowercase__    :	Dict     					=			prefix_result[j - 1]

              if input_string[i] == input_string[j]:
                     j += 1
              lowercase__    :	Union[str, Any]     					=			j

       return prefix_result





def        a_  (   _lowerCAmelCase						:				str  ):





       '''simple docstring'''

       return max(prefix_function(_lowerCAmelCase  )  )


if __name__ == "__main__":
       import doctest

       doctest.testmod()

