



import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


__lowerCamelCase         =			logging.get_logger(__name__)  # pylint: disable=invalid-name


class   snake_case_ (lowercase__  ):




       """simple docstring"""






       def __init__(     self     ,lowercase     ,lowercase     ,lowercase     ,lowercase     ,lowercase     ,lowercase     ,lowercase     ,lowercase     ,lowercase     ,):

             """simple docstring"""





             super().__init__()

             if safety_checker is None:
                   logger.warning(
                       F"""You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"""
                       " that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
                       " results in services or applications open to the public. Both the diffusers team and Hugging Face"
                       " strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
                       " it only for use-cases that involve analyzing network behavior or auditing its results. For more"
                       " information, please have a look at https://github.com/huggingface/diffusers/pull/254 .")

             self.register_modules(
                 speech_model=lowercase     ,speech_processor=lowercase     ,vae=lowercase     ,text_encoder=lowercase     ,tokenizer=lowercase     ,unet=lowercase     ,scheduler=lowercase     ,feature_extractor=lowercase     ,)



       def  A_   (     self     ,lowercase = "auto"):

             """simple docstring"""





             if slice_size == "auto":
                   UpperCAmelCase_  :	List[str]				      =		self.unet.config.attention_head_dim // 2
             self.unet.set_attention_slice(lowercase)



       def  A_   (     self):

             """simple docstring"""





             self.enable_attention_slicing(lowercase)



       @torch.no_grad()
       def __call__(     self     ,lowercase     ,lowercase=16000     ,lowercase = 512     ,lowercase = 512     ,lowercase = 50     ,lowercase = 7.5     ,lowercase = None     ,lowercase = 1     ,lowercase = 0.0     ,lowercase = None     ,lowercase = None     ,lowercase = "pil"     ,lowercase = True     ,lowercase = None     ,lowercase = 1     ,**lowercase     ,):

             """simple docstring"""





             UpperCAmelCase_  :	Tuple				      =		self.speech_processor.feature_extractor(
                 lowercase     ,return_tensors="pt"     ,sampling_rate=lowercase).input_features.to(self.device)
             UpperCAmelCase_  :	int				      =		self.speech_model.generate(lowercase     ,max_length=480000)

             UpperCAmelCase_  :	Any				      =		self.speech_processor.tokenizer.batch_decode(lowercase     ,skip_special_tokens=lowercase     ,normalize=lowercase)[
                 0
             ]

             if isinstance(lowercase     ,lowercase):
                   UpperCAmelCase_  :	Any				      =		1
             elif isinstance(lowercase     ,lowercase):
                   UpperCAmelCase_  :	Optional[int]				      =		len(lowercase)
             else:
                   raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(lowercase)}""")

             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 (callback_steps is None) or (
                 callback_steps is not None and (not isinstance(lowercase     ,lowercase) or callback_steps <= 0)
             ):
                   raise ValueError(
                       F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
                       F""" {type(lowercase)}.""")

             # get prompt text embeddings
             UpperCAmelCase_  :	Any				      =		self.tokenizer(
                 lowercase     ,padding="max_length"     ,max_length=self.tokenizer.model_max_length     ,return_tensors="pt"     ,)
             UpperCAmelCase_  :	Tuple				      =		text_inputs.input_ids

             if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                   UpperCAmelCase_  :	Optional[Any]				      =		self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :])
                   logger.warning(
                       "The following part of your input was truncated because CLIP can only handle sequences up to"
                       F""" {self.tokenizer.model_max_length} tokens: {removed_text}""")
                   UpperCAmelCase_  :	Tuple				      =		text_input_ids[:, : self.tokenizer.model_max_length]
             UpperCAmelCase_  :	Optional[int]				      =		self.text_encoder(text_input_ids.to(self.device))[0]

             # duplicate text embeddings for each generation per prompt, using mps friendly method
             UpperCAmelCase_  :	List[Any]				      =		text_embeddings.shape
             UpperCAmelCase_  :	Dict				      =		text_embeddings.repeat(1     ,lowercase     ,1)
             UpperCAmelCase_  :	Union[str, Any]				      =		text_embeddings.view(bs_embed * num_images_per_prompt     ,lowercase     ,-1)

             # 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_  :	int				      =		guidance_scale > 1.0
             # get unconditional embeddings for classifier free guidance
             if do_classifier_free_guidance:
                   UpperCAmelCase_  :	List[str]
                   if negative_prompt is None:
                         UpperCAmelCase_  :	Any				      =		[""] * batch_size
                   elif type(lowercase) is not type(lowercase):
                         raise TypeError(
                             F"""`negative_prompt` should be the same type to `prompt`, but got {type(lowercase)} !="""
                             F""" {type(lowercase)}.""")
                   elif isinstance(lowercase     ,lowercase):
                         UpperCAmelCase_  :	Union[str, Any]				      =		[negative_prompt]
                   elif batch_size != len(lowercase):
                         raise ValueError(
                             F"""`negative_prompt`: {negative_prompt} has batch size {len(lowercase)}, but `prompt`:"""
                             F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                             " the batch size of `prompt`.")
                   else:
                         UpperCAmelCase_  :	int				      =		negative_prompt

                   UpperCAmelCase_  :	Dict				      =		text_input_ids.shape[-1]
                   UpperCAmelCase_  :	str				      =		self.tokenizer(
                       lowercase     ,padding="max_length"     ,max_length=lowercase     ,truncation=lowercase     ,return_tensors="pt"     ,)
                   UpperCAmelCase_  :	Optional[Any]				      =		self.text_encoder(uncond_input.input_ids.to(self.device))[0]

                   # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
                   UpperCAmelCase_  :	List[Any]				      =		uncond_embeddings.shape[1]
                   UpperCAmelCase_  :	Union[str, Any]				      =		uncond_embeddings.repeat(1     ,lowercase     ,1)
                   UpperCAmelCase_  :	Dict				      =		uncond_embeddings.view(batch_size * num_images_per_prompt     ,lowercase     ,-1)

                   # 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_  :	List[str]				      =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
             UpperCAmelCase_  :	Any				      =		text_embeddings.dtype
             if latents is None:
                   if self.device.type == "mps":
                         # randn does not exist on mps
                         UpperCAmelCase_  :	Optional[int]				      =		torch.randn(lowercase     ,generator=lowercase     ,device="cpu"     ,dtype=lowercase).to(
                             self.device)
                   else:
                         UpperCAmelCase_  :	List[str]				      =		torch.randn(lowercase     ,generator=lowercase     ,device=self.device     ,dtype=lowercase)
             else:
                   if latents.shape != latents_shape:
                         raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}""")
                   UpperCAmelCase_  :	Optional[int]				      =		latents.to(self.device)

             # set timesteps
             self.scheduler.set_timesteps(lowercase)

             # Some schedulers like PNDM have timesteps as arrays
             # It's more optimized to move all timesteps to correct device beforehand
             UpperCAmelCase_  :	Dict				      =		self.scheduler.timesteps.to(self.device)

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

             for i, t in enumerate(self.progress_bar(lowercase)):
                   # expand the latents if we are doing classifier free guidance
                   UpperCAmelCase_  :	Optional[int]				      =		torch.cat([latents] * 2) if do_classifier_free_guidance else latents
                   UpperCAmelCase_  :	Tuple				      =		self.scheduler.scale_model_input(lowercase     ,lowercase)

                   # predict the noise residual
                   UpperCAmelCase_  :	Optional[int]				      =		self.unet(lowercase     ,lowercase     ,encoder_hidden_states=lowercase).sample

                   # perform guidance
                   if do_classifier_free_guidance:
                         UpperCAmelCase_  :	int				      =		noise_pred.chunk(2)
                         UpperCAmelCase_  :	List[Any]				      =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                   # compute the previous noisy sample x_t -> x_t-1
                   UpperCAmelCase_  :	Union[str, Any]				      =		self.scheduler.step(lowercase     ,lowercase     ,lowercase     ,**lowercase).prev_sample

                   # call the callback, if provided
                   if callback is not None and i % callback_steps == 0:
                         callback(lowercase     ,lowercase     ,lowercase)

             UpperCAmelCase_  :	str				      =		1 / 0.1_8215 * latents
             UpperCAmelCase_  :	Any				      =		self.vae.decode(lowercase).sample

             UpperCAmelCase_  :	str				      =		(image / 2 + 0.5).clamp(0     ,1)

             # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
             UpperCAmelCase_  :	Tuple				      =		image.cpu().permute(0     ,2     ,3     ,1).float().numpy()

             if output_type == "pil":
                   UpperCAmelCase_  :	List[Any]				      =		self.numpy_to_pil(lowercase)

             if not return_dict:
                   return image

             return StableDiffusionPipelineOutput(images=lowercase     ,nsfw_content_detected=lowercase)





from ...configuration_utils import PretrainedConfig
from ...utils import logging


__lowerCamelCase         =			logging.get_logger(__name__)

__lowerCamelCase         =			{
    '''unc-nlp/lxmert-base-uncased''': '''https://huggingface.co/unc-nlp/lxmert-base-uncased/resolve/main/config.json''',
}


class   snake_case_ (lowercase__  ):




       """simple docstring"""






       _lowerCamelCase			    =				"""lxmert"""
       _lowerCamelCase			    =				{}



       def __init__(     self     ,lowercase=30522     ,lowercase=768     ,lowercase=12     ,lowercase=9500     ,lowercase=1600     ,lowercase=400     ,lowercase=3072     ,lowercase="gelu"     ,lowercase=0.1     ,lowercase=0.1     ,lowercase=512     ,lowercase=2     ,lowercase=0.02     ,lowercase=1E-12     ,lowercase=9     ,lowercase=5     ,lowercase=5     ,lowercase=2048     ,lowercase=4     ,lowercase=6.67     ,lowercase=True     ,lowercase=True     ,lowercase=True     ,lowercase=True     ,lowercase=True     ,lowercase=True     ,lowercase=True     ,**lowercase     ,):

             """simple docstring"""





             UpperCAmelCase_  :	List[str]				      =		vocab_size
             UpperCAmelCase_  :	str				      =		hidden_size
             UpperCAmelCase_  :	List[str]				      =		num_attention_heads
             UpperCAmelCase_  :	Dict				      =		hidden_act
             UpperCAmelCase_  :	List[Any]				      =		intermediate_size
             UpperCAmelCase_  :	Tuple				      =		hidden_dropout_prob
             UpperCAmelCase_  :	List[Any]				      =		attention_probs_dropout_prob
             UpperCAmelCase_  :	Any				      =		max_position_embeddings
             UpperCAmelCase_  :	str				      =		type_vocab_size
             UpperCAmelCase_  :	Any				      =		initializer_range
             UpperCAmelCase_  :	Optional[Any]				      =		layer_norm_eps
             UpperCAmelCase_  :	Any				      =		num_qa_labels
             UpperCAmelCase_  :	str				      =		num_object_labels
             UpperCAmelCase_  :	Dict				      =		num_attr_labels
             UpperCAmelCase_  :	Tuple				      =		l_layers
             UpperCAmelCase_  :	Tuple				      =		x_layers
             UpperCAmelCase_  :	int				      =		r_layers
             UpperCAmelCase_  :	Optional[Any]				      =		visual_feat_dim
             UpperCAmelCase_  :	List[Any]				      =		visual_pos_dim
             UpperCAmelCase_  :	int				      =		visual_loss_normalizer
             UpperCAmelCase_  :	str				      =		task_matched
             UpperCAmelCase_  :	str				      =		task_mask_lm
             UpperCAmelCase_  :	int				      =		task_obj_predict
             UpperCAmelCase_  :	List[str]				      =		task_qa
             UpperCAmelCase_  :	Optional[int]				      =		visual_obj_loss
             UpperCAmelCase_  :	List[str]				      =		visual_attr_loss
             UpperCAmelCase_  :	str				      =		visual_feat_loss
             UpperCAmelCase_  :	List[Any]				      =		{"vision": r_layers, "cross_encoder": x_layers, "language": l_layers}
             super().__init__(**lowercase)
