'''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  _SCREAMING_SNAKE_CASE		(      __snake_case							:			List[str]   ,		__snake_case							:			Optional[int]   ,		__snake_case							:			List[str]				):
     if isinstance(__snake_case   ,		torch.Tensor				):
          return image
     elif isinstance(__snake_case   ,		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(__snake_case   ,		axis=0				)
          _A    		=  np.array(__snake_case				).astype(np.floataa				) / 2_55.0
          _A    		=  image.transpose(0   ,		3   ,		1   ,		2				)
          _A    		=  2.0 * image - 1.0
          _A    		=  torch.from_numpy(__snake_case				)
     elif isinstance(image[0]   ,		torch.Tensor				):
          _A    		=  torch.cat(__snake_case   ,		dim=0				)
     return image





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			List[Any]   ,		__snake_case							:			List[str]   ,		__snake_case							:			Optional[Any]   ,		__snake_case							:			Optional[Any]=0.99_95				):
     if not isinstance(__snake_case   ,		np.ndarray				):
          _A    		=  True
          _A    		=  va.device
          _A    		=  va.cpu().numpy()
          _A    		=  va.cpu().numpy()

     _A    		=  np.sum(va * va / (np.linalg.norm(__snake_case				) * np.linalg.norm(__snake_case				))				)
     if np.abs(__snake_case				) > DOT_THRESHOLD:
          _A    		=  (1 - t) * va + t * va
     else:
          _A    		=  np.arccos(__snake_case				)
          _A    		=  np.sin(__snake_case				)
          _A    		=  theta_a * t
          _A    		=  np.sin(__snake_case				)
          _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(__snake_case				).to(__snake_case				)

     return va





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			Union[str, Any]   ,		__snake_case							:			int				):
     _A    		=  F.normalize(__snake_case   ,		dim=-1				)
     _A    		=  F.normalize(__snake_case   ,		dim=-1				)
     return (x - y).norm(dim=-1				).div(2				).arcsin().pow(2				).mul(2				)





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			Dict   ,		__snake_case							:			str				):
     for param in model.parameters():
          _A    		=  value







class lowercase_       (		_UpperCamelCase			):







   """simple docstring"""

   def __init__(      self      :			List[str], UpperCamelCase__      :			AutoencoderKL, UpperCamelCase__      :			CLIPTextModel, UpperCamelCase__      :			CLIPModel, UpperCamelCase__      :			CLIPTokenizer, UpperCamelCase__      :			UNetaDConditionModel, UpperCamelCase__      :			Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler], UpperCamelCase__      :			CLIPFeatureExtractor, UpperCamelCase__      :			Any=None, UpperCamelCase__      :			Optional[Any]=None, UpperCamelCase__      :			Tuple=None, )  ->			Optional[Any]:
        super().__init__()
        self.register_modules(
            vae=UpperCamelCase__, text_encoder=UpperCamelCase__, clip_model=UpperCamelCase__, tokenizer=UpperCamelCase__, unet=UpperCamelCase__, scheduler=UpperCamelCase__, feature_extractor=UpperCamelCase__, coca_model=UpperCamelCase__, coca_tokenizer=UpperCamelCase__, coca_transform=UpperCamelCase__, )
        _A    		=  (
            feature_extractor.size
            if isinstance(feature_extractor.size, UpperCamelCase__     )
            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, UpperCamelCase__     )
        set_requires_grad(self.clip_model, UpperCamelCase__     )
   def 	__UpperCAmelCase   (      self      :			List[Any], UpperCamelCase__      :			Optional[Union[str, int]] = "auto"     )  ->			Optional[Any]:
        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(UpperCamelCase__     )
   def 	__UpperCAmelCase   (      self      :			Optional[Any]     )  ->			Tuple:
        self.enable_attention_slicing(UpperCamelCase__     )
   def 	__UpperCAmelCase   (      self      :			Optional[Any]     )  ->			Union[str, Any]:
        set_requires_grad(self.vae, UpperCamelCase__     )
   def 	__UpperCAmelCase   (      self      :			Union[str, Any]     )  ->			str:
        set_requires_grad(self.vae, UpperCamelCase__     )
   def 	__UpperCAmelCase   (      self      :			List[str]     )  ->			int:
        set_requires_grad(self.unet, UpperCamelCase__     )
   def 	__UpperCAmelCase   (      self      :			List[str]     )  ->			Any:
        set_requires_grad(self.unet, UpperCamelCase__     )
   def 	__UpperCAmelCase   (      self      :			Union[str, Any], UpperCamelCase__      :			Optional[int], UpperCamelCase__      :			List[str], UpperCamelCase__      :			Any     )  ->			Dict:
        # get the original timestep using init_timestep
        _A    		=  min(int(num_inference_steps * strength     ), UpperCamelCase__     )

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

        return timesteps, num_inference_steps - t_start
   def 	__UpperCAmelCase   (      self      :			List[Any], UpperCamelCase__      :			Optional[Any], UpperCamelCase__      :			str, UpperCamelCase__      :			Tuple, UpperCamelCase__      :			Any, UpperCamelCase__      :			List[str], UpperCamelCase__      :			List[Any]=None     )  ->			Any:
        if not isinstance(UpperCamelCase__, torch.Tensor     ):
             raise ValueError(f'`image` has to be of type `torch.Tensor` but is {type(UpperCamelCase__     )}'     )

        _A    		=  image.to(device=UpperCamelCase__, dtype=UpperCamelCase__     )

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

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

        _A    		=  randn_tensor(init_latents.shape, generator=UpperCamelCase__, device=UpperCamelCase__, dtype=UpperCamelCase__     )

        # get latents
        _A    		=  self.scheduler.add_noise(UpperCamelCase__, UpperCamelCase__, UpperCamelCase__     )
        _A    		=  init_latents

        return latents
   def 	__UpperCAmelCase   (      self      :			int, UpperCamelCase__      :			Optional[int]     )  ->			List[str]:
        _A    		=  self.coca_transform(UpperCamelCase__     ).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 	__UpperCAmelCase   (      self      :			Union[str, Any], UpperCamelCase__      :			List[Any], UpperCamelCase__      :			Optional[Any]     )  ->			List[str]:
        _A    		=  self.feature_extractor.preprocess(UpperCamelCase__     )
        _A    		=  torch.from_numpy(clip_image_input['pixel_values'][0]     ).unsqueeze(0     ).to(self.device     ).half()
        _A    		=  self.clip_model.get_image_features(UpperCamelCase__     )
        _A    		=  image_embeddings_clip / image_embeddings_clip.norm(p=2, dim=-1, keepdim=UpperCamelCase__     )
        _A    		=  image_embeddings_clip.repeat_interleave(UpperCamelCase__, dim=0     )
        return image_embeddings_clip
   @torch.enable_grad()
   def 	__UpperCAmelCase   (      self      :			List[Any], UpperCamelCase__      :			Any, UpperCamelCase__      :			Union[str, Any], UpperCamelCase__      :			Dict, UpperCamelCase__      :			Tuple, UpperCamelCase__      :			int, UpperCamelCase__      :			List[str], UpperCamelCase__      :			List[Any], )  ->			int:
        _A    		=  latents.detach().requires_grad_()

        _A    		=  self.scheduler.scale_model_input(UpperCamelCase__, UpperCamelCase__     )

        # predict the noise residual
        _A    		=  self.unet(UpperCamelCase__, UpperCamelCase__, encoder_hidden_states=UpperCamelCase__     ).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(UpperCamelCase__     )
             _A    		=  pred_original_sample * (fac) + latents * (1 - fac)
        elif isinstance(self.scheduler, UpperCamelCase__     ):
             _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.18_215 * sample
        _A    		=  self.vae.decode(UpperCamelCase__     ).sample
        _A    		=  (image / 2 + 0.5).clamp(0, 1     )

        _A    		=  transforms.Resize(self.feature_extractor_size     )(UpperCamelCase__     )
        _A    		=  self.normalize(UpperCamelCase__     ).to(latents.dtype     )

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

        _A    		=  spherical_dist_loss(UpperCamelCase__, UpperCamelCase__     ).mean() * clip_guidance_scale

        _A    		=  -torch.autograd.grad(UpperCamelCase__, UpperCamelCase__     )[0]

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

   @torch.no_grad()
   def __call__(      self      :			Optional[Any], UpperCamelCase__      :			Union[torch.FloatTensor, PIL.Image.Image], UpperCamelCase__      :			Union[torch.FloatTensor, PIL.Image.Image], UpperCamelCase__      :			Optional[str] = None, UpperCamelCase__      :			Optional[str] = None, UpperCamelCase__      :			Optional[int] = 5_12, UpperCamelCase__      :			Optional[int] = 5_12, UpperCamelCase__      :			float = 0.6, UpperCamelCase__      :			Optional[int] = 50, UpperCamelCase__      :			Optional[float] = 7.5, UpperCamelCase__      :			Optional[int] = 1, UpperCamelCase__      :			float = 0.0, UpperCamelCase__      :			Optional[float] = 1_00, UpperCamelCase__      :			Optional[torch.Generator] = None, UpperCamelCase__      :			Optional[str] = "pil", UpperCamelCase__      :			bool = True, UpperCamelCase__      :			float = 0.8, UpperCamelCase__      :			float = 0.1, UpperCamelCase__      :			float = 0.1, )  ->			Tuple:
        if isinstance(UpperCamelCase__, UpperCamelCase__     ) and len(UpperCamelCase__     ) != batch_size:
             raise ValueError(f'You have passed {batch_size} batch_size, but only {len(UpperCamelCase__     )} 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(UpperCamelCase__, 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(UpperCamelCase__     )
        # generate prompts with coca model if prompt is None
        if content_prompt is None:
             if len(UpperCamelCase__     ):
                  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(UpperCamelCase__     )
        if style_prompt is None:
             if len(UpperCamelCase__     ):
                  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(UpperCamelCase__     )

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

        _A    		=  self.tokenizer(
            UpperCamelCase__, padding='max_length', max_length=self.tokenizer.model_max_length, truncation=UpperCamelCase__, return_tensors='pt', )
        _A    		=  self.text_encoder(style_text_input.input_ids.to(self.device     )     )[0]

        _A    		=  slerp(UpperCamelCase__, UpperCamelCase__, UpperCamelCase__     )

        # duplicate text embeddings for each generation per prompt
        _A    		=  text_embeddings.repeat_interleave(UpperCamelCase__, 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(UpperCamelCase__, **UpperCamelCase__     )
        # 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(UpperCamelCase__, UpperCamelCase__, self.device     )
        _A    		=  timesteps[:1].repeat(UpperCamelCase__     )

        # Preprocess image
        _A    		=  preprocess(UpperCamelCase__, UpperCamelCase__, UpperCamelCase__     )
        _A    		=  self.prepare_latents(
            UpperCamelCase__, UpperCamelCase__, UpperCamelCase__, text_embeddings.dtype, self.device, UpperCamelCase__     )

        _A    		=  preprocess(UpperCamelCase__, UpperCamelCase__, UpperCamelCase__     )
        _A    		=  self.prepare_latents(
            UpperCamelCase__, UpperCamelCase__, UpperCamelCase__, text_embeddings.dtype, self.device, UpperCamelCase__     )

        _A    		=  slerp(UpperCamelCase__, UpperCamelCase__, UpperCamelCase__     )

        if clip_guidance_scale > 0:
             _A    		=  self.get_clip_image_embeddings(UpperCamelCase__, UpperCamelCase__     )
             _A    		=  self.get_clip_image_embeddings(UpperCamelCase__, UpperCamelCase__     )
             _A    		=  slerp(
                 UpperCamelCase__, UpperCamelCase__, UpperCamelCase__     )

        # 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=UpperCamelCase__, 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(UpperCamelCase__, 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(UpperCamelCase__, generator=UpperCamelCase__, device='cpu', dtype=UpperCamelCase__     ).to(
                      self.device     )
             else:
                  _A    		=  torch.randn(UpperCamelCase__, generator=UpperCamelCase__, device=self.device, dtype=UpperCamelCase__     )
        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=UpperCamelCase__     ):
             for i, t in enumerate(UpperCamelCase__     ):
                  # 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(UpperCamelCase__, UpperCamelCase__     )

                  # predict the noise residual
                  _A    		=  self.unet(UpperCamelCase__, UpperCamelCase__, encoder_hidden_states=UpperCamelCase__     ).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(
                           UpperCamelCase__, UpperCamelCase__, UpperCamelCase__, UpperCamelCase__, UpperCamelCase__, UpperCamelCase__, UpperCamelCase__, )

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

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
        _A    		=  1 / 0.18_215 * latents
        _A    		=  self.vae.decode(UpperCamelCase__     ).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(UpperCamelCase__     )

        if not return_dict:
             return (image, None)

        return StableDiffusionPipelineOutput(images=UpperCamelCase__, nsfw_content_detected=UpperCamelCase__     )







def    a__    (		snake_case__					:							int						,   snake_case__					:							int     ):
  return 1 if input_a == input_a else 0




def    a__    (		):
  assert xnor_gate(0						,   0     ) == 1
  assert xnor_gate(0						,   1     ) == 0
  assert xnor_gate(1						,   0     ) == 0
  assert xnor_gate(1						,   1     ) == 1


if __name__ == "__main__":
 print(xnor_gate(0, 0))
 print(xnor_gate(0, 1))
 print(xnor_gate(1, 0))
 print(xnor_gate(1, 1))





