






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							(					__lowerCAmelCase					,    __lowerCAmelCase					,    __lowerCAmelCase			) ->	List[str]:






     if isinstance(__a					,    torch.Tensor			):
          return image
     elif isinstance(__a					,    PIL.Image.Image			):
          UpperCamelCase__			:   Tuple       				=  [image]

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







def 		SCREAMING_SNAKE_CASE							(					__lowerCAmelCase					,    __lowerCAmelCase					,    __lowerCAmelCase					,    __lowerCAmelCase=0.9_9_9_5			) ->	Dict:






     if not isinstance(__a					,    np.ndarray			):
          UpperCamelCase__			:   str       				=  True
          UpperCamelCase__			:   Optional[int]       				=  va.device
          UpperCamelCase__			:   List[str]       				=  va.cpu().numpy()
          UpperCamelCase__			:   int       				=  va.cpu().numpy()

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

     if inputs_are_torch:
          UpperCamelCase__			:   List[str]       				=  torch.from_numpy(__a			).to(__a			)

     return va







def 		SCREAMING_SNAKE_CASE							(					__lowerCAmelCase					,    __lowerCAmelCase			) ->	int:
     UpperCamelCase__			:   str       				=  F.normalize(__a					,    dim=-1			)
     UpperCamelCase__			:   Any       				=  F.normalize(__a					,    dim=-1			)
     return (x - y).norm(dim=-1			).div(2			).arcsin().pow(2			).mul(2			)







def 		SCREAMING_SNAKE_CASE							(					__lowerCAmelCase					,    __lowerCAmelCase			) ->	Any:






     for param in model.parameters():
          UpperCamelCase__			:   Any       				=  value





class 			__a  (    A__       ):







 def __init__(		self					:   str							,   SCREAMING_SNAKE_CASE					:   str							,   SCREAMING_SNAKE_CASE					:   List[str]							,   SCREAMING_SNAKE_CASE					:   Optional[int]							,   SCREAMING_SNAKE_CASE					:   Optional[int]							,   SCREAMING_SNAKE_CASE					:   int							,   SCREAMING_SNAKE_CASE					:   Any							,   SCREAMING_SNAKE_CASE					:   Union[str, Any]							,   SCREAMING_SNAKE_CASE					:   List[str]=None							,   SCREAMING_SNAKE_CASE					:   str=None							,   SCREAMING_SNAKE_CASE					:   List[Any]=None							,   ):


      '''simple docstring'''






      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_							,   )
      UpperCamelCase__			:   List[Any]       				=  (
          feature_extractor.size
          if isinstance(feature_extractor.size							,   UpperCamelCase_						)
          else feature_extractor.size['''shortest_edge''']
      )
      UpperCamelCase__			:   List[str]       				=  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 					__lowercase					(		self					:   List[Any]							,   SCREAMING_SNAKE_CASE					:   Dict = "auto"						):


      '''simple docstring'''






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







 def 					__lowercase					(		self					:   int						):


      '''simple docstring'''






      self.enable_attention_slicing(UpperCamelCase_						)







 def 					__lowercase					(		self					:   List[str]						):


      '''simple docstring'''






      set_requires_grad(self.vae							,   UpperCamelCase_						)







 def 					__lowercase					(		self					:   List[Any]						):


      '''simple docstring'''






      set_requires_grad(self.vae							,   UpperCamelCase_						)







 def 					__lowercase					(		self					:   List[str]						):


      '''simple docstring'''






      set_requires_grad(self.unet							,   UpperCamelCase_						)







 def 					__lowercase					(		self					:   Optional[int]						):


      '''simple docstring'''






      set_requires_grad(self.unet							,   UpperCamelCase_						)







 def 					__lowercase					(		self					:   Union[str, Any]							,   SCREAMING_SNAKE_CASE					:   Optional[int]							,   SCREAMING_SNAKE_CASE					:   Optional[Any]							,   SCREAMING_SNAKE_CASE					:   Any						):


      '''simple docstring'''






      UpperCamelCase__			:   int       				=  min(int(num_inference_steps * strength						)							,   UpperCamelCase_						)

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

      return timesteps, num_inference_steps - t_start







 def 					__lowercase					(		self					:   str							,   SCREAMING_SNAKE_CASE					:   str							,   SCREAMING_SNAKE_CASE					:   Tuple							,   SCREAMING_SNAKE_CASE					:   Dict							,   SCREAMING_SNAKE_CASE					:   List[Any]							,   SCREAMING_SNAKE_CASE					:   Optional[int]							,   SCREAMING_SNAKE_CASE					:   List[Any]=None						):


      '''simple docstring'''






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

      UpperCamelCase__			:   List[str]       				=  image.to(device=UpperCamelCase_							,   dtype=UpperCamelCase_						)

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

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

      UpperCamelCase__			:   Any       				=  randn_tensor(init_latents.shape							,   generator=UpperCamelCase_							,   device=UpperCamelCase_							,   dtype=UpperCamelCase_						)

      # get latents
      UpperCamelCase__			:   Tuple       				=  self.scheduler.add_noise(UpperCamelCase_							,   UpperCamelCase_							,   UpperCamelCase_						)
      UpperCamelCase__			:   List[Any]       				=  init_latents

      return latents







 def 					__lowercase					(		self					:   str							,   SCREAMING_SNAKE_CASE					:   Optional[Any]						):


      '''simple docstring'''






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







 def 					__lowercase					(		self					:   Tuple							,   SCREAMING_SNAKE_CASE					:   int							,   SCREAMING_SNAKE_CASE					:   Dict						):


      '''simple docstring'''






      UpperCamelCase__			:   Union[str, Any]       				=  self.feature_extractor.preprocess(UpperCamelCase_						)
      UpperCamelCase__			:   Any       				=  torch.from_numpy(clip_image_input["pixel_values"][0]						).unsqueeze(0						).to(self.device						).half()
      UpperCamelCase__			:   str       				=  self.clip_model.get_image_features(UpperCamelCase_						)
      UpperCamelCase__			:   List[str]       				=  image_embeddings_clip / image_embeddings_clip.norm(p=2							,   dim=-1							,   keepdim=UpperCamelCase_						)
      UpperCamelCase__			:   Union[str, Any]       				=  image_embeddings_clip.repeat_interleave(UpperCamelCase_							,   dim=0						)
      return image_embeddings_clip







 @torch.enable_grad()
 def 					__lowercase					(		self					:   Any							,   SCREAMING_SNAKE_CASE					:   str							,   SCREAMING_SNAKE_CASE					:   List[str]							,   SCREAMING_SNAKE_CASE					:   Any							,   SCREAMING_SNAKE_CASE					:   Any							,   SCREAMING_SNAKE_CASE					:   List[str]							,   SCREAMING_SNAKE_CASE					:   str							,   SCREAMING_SNAKE_CASE					:   Tuple							,   ):


      '''simple docstring'''






      UpperCamelCase__			:   List[str]       				=  latents.detach().requires_grad_()

      UpperCamelCase__			:   List[Any]       				=  self.scheduler.scale_model_input(UpperCamelCase_							,   UpperCamelCase_						)

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

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

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

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

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

      UpperCamelCase__			:   Any       				=  spherical_dist_loss(UpperCamelCase_							,   UpperCamelCase_						).mean() * clip_guidance_scale

      UpperCamelCase__			:   str       				=  -torch.autograd.grad(UpperCamelCase_							,   UpperCamelCase_						)[0]

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







 @torch.no_grad()
 def __call__(		self					:   str							,   SCREAMING_SNAKE_CASE					:   Dict							,   SCREAMING_SNAKE_CASE					:   List[Any]							,   SCREAMING_SNAKE_CASE					:   Dict = None							,   SCREAMING_SNAKE_CASE					:   Dict = None							,   SCREAMING_SNAKE_CASE					:   Union[str, Any] = 5_12							,   SCREAMING_SNAKE_CASE					:   List[str] = 5_12							,   SCREAMING_SNAKE_CASE					:   Optional[int] = 0.6							,   SCREAMING_SNAKE_CASE					:   Any = 50							,   SCREAMING_SNAKE_CASE					:   Dict = 7.5							,   SCREAMING_SNAKE_CASE					:   Union[str, Any] = 1							,   SCREAMING_SNAKE_CASE					:   List[Any] = 0.0							,   SCREAMING_SNAKE_CASE					:   str = 1_00							,   SCREAMING_SNAKE_CASE					:   Dict = None							,   SCREAMING_SNAKE_CASE					:   int = "pil"							,   SCREAMING_SNAKE_CASE					:   Union[str, Any] = True							,   SCREAMING_SNAKE_CASE					:   Any = 0.8							,   SCREAMING_SNAKE_CASE					:   Dict = 0.1							,   SCREAMING_SNAKE_CASE					:   str = 0.1							,   ):


      '''simple docstring'''






      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:
           UpperCamelCase__			:   List[Any]       				=  [generator] + [None] * (batch_size - 1)

      UpperCamelCase__			:   Union[str, Any]       				=  [
          ('''model''', self.coca_model is None),
          ('''tokenizer''', self.coca_tokenizer is None),
          ('''transform''', self.coca_transform is None),
      ]
      UpperCamelCase__			:   Union[str, Any]       				=  [x[0] for x in coca_is_none if x[1]]
      UpperCamelCase__			:   Union[str, Any]       				=  ''', '''.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.'						)
           UpperCamelCase__			:   List[str]       				=  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.'						)
           UpperCamelCase__			:   List[Any]       				=  self.get_image_description(UpperCamelCase_						)

      # get prompt text embeddings for content and style
      UpperCamelCase__			:   Union[str, Any]       				=  self.tokenizer(
          UpperCamelCase_							,   padding="max_length"							,   max_length=self.tokenizer.model_max_length							,   truncation=UpperCamelCase_							,   return_tensors="pt"							,   )
      UpperCamelCase__			:   Any       				=  self.text_encoder(content_text_input.input_ids.to(self.device						)						)[0]

      UpperCamelCase__			:   Any       				=  self.tokenizer(
          UpperCamelCase_							,   padding="max_length"							,   max_length=self.tokenizer.model_max_length							,   truncation=UpperCamelCase_							,   return_tensors="pt"							,   )
      UpperCamelCase__			:   Tuple       				=  self.text_encoder(style_text_input.input_ids.to(self.device						)						)[0]

      UpperCamelCase__			:   str       				=  slerp(UpperCamelCase_							,   UpperCamelCase_							,   UpperCamelCase_						)

      # duplicate text embeddings for each generation per prompt
      UpperCamelCase__			:   List[str]       				=  text_embeddings.repeat_interleave(UpperCamelCase_							,   dim=0						)

      # set timesteps
      UpperCamelCase__			:   Dict       				=  '''offset''' in set(inspect.signature(self.scheduler.set_timesteps						).parameters.keys()						)
      UpperCamelCase__			:   Any       				=  {}
      if accepts_offset:
           UpperCamelCase__			:   Tuple       				=  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						)

      UpperCamelCase__			:   int       				=  self.get_timesteps(UpperCamelCase_							,   UpperCamelCase_							,   self.device						)
      UpperCamelCase__			:   Optional[Any]       				=  timesteps[:1].repeat(UpperCamelCase_						)

      # Preprocess image
      UpperCamelCase__			:   Any       				=  preprocess(UpperCamelCase_							,   UpperCamelCase_							,   UpperCamelCase_						)
      UpperCamelCase__			:   Dict       				=  self.prepare_latents(
          UpperCamelCase_							,   UpperCamelCase_							,   UpperCamelCase_							,   text_embeddings.dtype							,   self.device							,   UpperCamelCase_						)

      UpperCamelCase__			:   Optional[int]       				=  preprocess(UpperCamelCase_							,   UpperCamelCase_							,   UpperCamelCase_						)
      UpperCamelCase__			:   Optional[Any]       				=  self.prepare_latents(
          UpperCamelCase_							,   UpperCamelCase_							,   UpperCamelCase_							,   text_embeddings.dtype							,   self.device							,   UpperCamelCase_						)

      UpperCamelCase__			:   int       				=  slerp(UpperCamelCase_							,   UpperCamelCase_							,   UpperCamelCase_						)

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

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

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

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

                # predict the noise residual
                UpperCamelCase__			:   str       				=  self.unet(UpperCamelCase_							,   UpperCamelCase_							,   encoder_hidden_states=UpperCamelCase_						).sample

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

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

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

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      UpperCamelCase__			:   Any       				=  1 / 0.1_8_2_1_5 * latents
      UpperCamelCase__			:   Union[str, Any]       				=  self.vae.decode(UpperCamelCase_						).sample

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

      if output_type == "pil":
           UpperCamelCase__			:   Tuple       				=  self.numpy_to_pil(UpperCamelCase_						)

      if not return_dict:
           return (image, None)

      return StableDiffusionPipelineOutput(images=UpperCamelCase_							,   nsfw_content_detected=UpperCamelCase_						)



'''simple docstring'''
from collections import defaultdict
class    lowercase	:

		"""simple docstring"""

		def __init__(		self    ,   UpperCamelCase_    ,   UpperCamelCase_							):

							'''simple docstring'''


							UpperCamelCase__   :List[Any]												= total  # total no of tasks (N)

							# DP table will have a dimension of (2^M)*N
							# initially all values are set to -1
							UpperCamelCase__   :Union[str, Any]												= [
							    [-1 for i in range(total + 1							)] for j in range(2 ** len(UpperCamelCase_							)							)
							]

							UpperCamelCase__   :str												= defaultdict(UpperCamelCase_							)  # stores the list of persons for each task

							# final_mask is used to check if all persons are included by setting all bits
							# to 1
							UpperCamelCase__   :Optional[int]												= (1 << len(UpperCamelCase_							)) - 1




		def     lowerCAmelCase__				(		self    ,   UpperCamelCase_    ,   UpperCamelCase_							):

							'''simple docstring'''


							if mask == self.final_mask:
												return 1

							# if not everyone gets the task and no more tasks are available, return 0
							if task_no > self.total_tasks:
												return 0

							# if case already considered
							if self.dp[mask][task_no] != -1:
												return self.dp[mask][task_no]

							# Number of ways when we don't this task in the arrangement
							UpperCamelCase__   :str												= self.count_ways_until(UpperCamelCase_    ,   task_no + 1							)

							# now assign the tasks one by one to all possible persons and recursively
							# assign for the remaining tasks.
							if task_no in self.task:
												for p in self.task[task_no]:
																	# if p is already given a task
																	if mask & (1 << p):
																						continue

																	# assign this task to p and change the mask value. And recursively
																	# assign tasks with the new mask value.
																	total_ways_util += self.count_ways_until(mask | (1 << p)    ,   task_no + 1							)

        # save the value.
							UpperCamelCase__   :Optional[int]												= total_ways_util

							return self.dp[mask][task_no]




		def     lowerCAmelCase__				(		self    ,   UpperCamelCase_							):

							'''simple docstring'''


							for i in range(len(UpperCamelCase_							)							):
												for j in task_performed[i]:
																	self.task[j].append(UpperCamelCase_							)

        # call the function to fill the DP table, final answer is stored in dp[0][1]
							return self.count_ways_until(0    ,   1							)


if __name__ == "__main__":
		__snake_case							      =     5  # total no of tasks (the value of N)

		# the list of tasks that can be done by M persons.
		__snake_case							      =     [[1, 3, 4], [1, 2, 5], [3, 4]]
		print(
		    AssignmentUsingBitmask(task_performed, total_tasks).count_no_of_ways(
		        task_performed
		    )
		)