






"""simple docstring"""




import inspect
import re

from transformers.utils import direct_transformers_import


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_config_docstrings.py
UpperCamelCase							       =    """src/transformers"""


# This is to make sure the transformers module imported is the one in the repo.
UpperCamelCase							       =    direct_transformers_import(PATH_TO_TRANSFORMERS)

UpperCamelCase							       =    transformers.models.auto.configuration_auto.CONFIG_MAPPING

# Regex pattern used to find the checkpoint mentioned in the docstring of `config_class`.
# For example, `[bert-base-uncased](https://huggingface.co/bert-base-uncased)`
UpperCamelCase							       =    re.compile(r"""\[(.+?)\]\((https://huggingface\.co/.+?)\)""")


UpperCamelCase							       =    {
    """DecisionTransformerConfig""",
    """EncoderDecoderConfig""",
    """MusicgenConfig""",
    """RagConfig""",
    """SpeechEncoderDecoderConfig""",
    """TimmBackboneConfig""",
    """VisionEncoderDecoderConfig""",
    """VisionTextDualEncoderConfig""",
    """LlamaConfig""",
}






def        lowerCAmelCase (      UpperCamelCase_:							List[str]					)       ->							Union[str, Any]:
  _a							=	None

  # source code of `config_class`
  _a							=	inspect.getsource(UpperCamelCase_					)
  _a							=	_re_checkpoint.findall(UpperCamelCase_					)

  # Each `checkpoint` is a tuple of a checkpoint name and a checkpoint link.
  # For example, `('bert-base-uncased', 'https://huggingface.co/bert-base-uncased')`
  for ckpt_name, ckpt_link in checkpoints:
    # allow the link to end with `/`
    if ckpt_link.endswith("/"					):
      _a							=	ckpt_link[:-1]

    # verify the checkpoint name corresponds to the checkpoint link
    _a							=	f'''https://huggingface.co/{ckpt_name}'''
    if ckpt_link == ckpt_link_from_name:
      _a							=	ckpt_name
      break

  return checkpoint






def        lowerCAmelCase (      )       ->							int:
  _a							=	[]

  for config_class in list(CONFIG_MAPPING.values()					):
    # Skip deprecated models
    if "models.deprecated" in config_class.__module__:
      continue
    _a							=	get_checkpoint_from_config_class(UpperCamelCase_					)

    _a							=	config_class.__name__
    if checkpoint is None and name not in CONFIG_CLASSES_TO_IGNORE_FOR_DOCSTRING_CHECKPOINT_CHECK:
      configs_without_checkpoint.append(UpperCamelCase_					)

  if len(UpperCamelCase_					) > 0:
    _a							=	"\n".join(sorted(UpperCamelCase_					)					)
    raise ValueError(f'''The following configurations don\'t contain any valid checkpoint:\n{message}'''					)


if __name__ == "__main__":
     check_config_docstrings_have_checkpoints()










"""simple docstring"""




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


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




class    lowercase_    (_UpperCAmelCase     ):






      def __init__(    self     ,     a_     ,     a_     ,     a_     ,     a_     ,     a_     ,     a_     ,     a_     ,     a_     ,     a_     ,     )	->Optional[Any]:







        '''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=a_     ,     speech_processor=a_     ,     vae=a_     ,     text_encoder=a_     ,     tokenizer=a_     ,     unet=a_     ,     scheduler=a_     ,     feature_extractor=a_     ,     )






      def 			lowerCamelCase__    (    self     ,     a_ = "auto"	)	->Optional[int]:







        '''simple docstring'''


        if slice_size == "auto":
          _a							=	self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(a_	)






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        self.enable_attention_slicing(a_	)






      @torch.no_grad()
      def __call__(    self     ,     a_     ,     a_=1_6_0_0_0     ,     a_ = 5_1_2     ,     a_ = 5_1_2     ,     a_ = 5_0     ,     a_ = 7.5     ,     a_ = None     ,     a_ = 1     ,     a_ = 0.0     ,     a_ = None     ,     a_ = None     ,     a_ = "pil"     ,     a_ = True     ,     a_ = None     ,     a_ = 1     ,     **a_     ,     )	->str:







        '''simple docstring'''


        _a							=	self.speech_processor.feature_extractor(
            a_     ,     return_tensors="pt"     ,     sampling_rate=a_	).input_features.to(self.device	)
        _a							=	self.speech_model.generate(a_     ,     max_length=4_8_0_0_0_0	)

        _a							=	self.speech_processor.tokenizer.batch_decode(a_     ,     skip_special_tokens=a_     ,     normalize=a_	)[
            0
        ]

        if isinstance(a_     ,     a_	):
          _a							=	1
        elif isinstance(a_     ,     a_	):
          _a							=	len(a_	)
        else:
          raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(a_	)}'''	)

        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(a_     ,     a_	) or callback_steps <= 0)
        ):
          raise ValueError(
              f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
              f''' {type(a_	)}.'''	)

        # get prompt text embeddings
        _a							=	self.tokenizer(
            a_     ,     padding="max_length"     ,     max_length=self.tokenizer.model_max_length     ,     return_tensors="pt"     ,     )
        _a							=	text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
          _a							=	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}'''	)
          _a							=	text_input_ids[:, : self.tokenizer.model_max_length]
        _a							=	self.text_encoder(text_input_ids.to(self.device	)	)[0]

        # duplicate text embeddings for each generation per prompt, using mps friendly method
        _a			,	_a			,	_a							=	text_embeddings.shape
        _a							=	text_embeddings.repeat(1     ,     a_     ,     1	)
        _a							=	text_embeddings.view(bs_embed * num_images_per_prompt     ,     a_     ,     -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.
        _a							=	guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
          _a							=	42
          if negative_prompt is None:
            _a							=	[""] * batch_size
          elif type(a_	) is not type(a_	):
            raise TypeError(
                f'''`negative_prompt` should be the same type to `prompt`, but got {type(a_	)} !='''
                f''' {type(a_	)}.'''	)
          elif isinstance(a_     ,     a_	):
            _a							=	[negative_prompt]
          elif batch_size != len(a_	):
            raise ValueError(
                f'''`negative_prompt`: {negative_prompt} has batch size {len(a_	)}, but `prompt`:'''
                f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                " the batch size of `prompt`."	)
          else:
            _a							=	negative_prompt

          _a							=	text_input_ids.shape[-1]
          _a							=	self.tokenizer(
              a_     ,     padding="max_length"     ,     max_length=a_     ,     truncation=a_     ,     return_tensors="pt"     ,     )
          _a							=	self.text_encoder(uncond_input.input_ids.to(self.device	)	)[0]

          # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
          _a							=	uncond_embeddings.shape[1]
          _a							=	uncond_embeddings.repeat(1     ,     a_     ,     1	)
          _a							=	uncond_embeddings.view(batch_size * num_images_per_prompt     ,     a_     ,     -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
          _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 * num_images_per_prompt, 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 exist on mps
            _a							=	torch.randn(a_     ,     generator=a_     ,     device="cpu"     ,     dtype=a_	).to(
                self.device	)
          else:
            _a							=	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}'''	)
          _a							=	latents.to(self.device	)

        # set timesteps
        self.scheduler.set_timesteps(a_	)

        # Some schedulers like PNDM have timesteps as arrays
        # It's more optimized to move all timesteps to correct device beforehand
        _a							=	self.scheduler.timesteps.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

        for i, t in enumerate(self.progress_bar(a_	)	):
          # 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(a_     ,     a_	)

          # predict the noise residual
          _a							=	self.unet(a_     ,     a_     ,     encoder_hidden_states=a_	).sample

          # perform guidance
          if do_classifier_free_guidance:
            _a			,	_a							=	noise_pred.chunk(2	)
            _a							=	noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

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

        _a							=	1 / 0.18_215 * latents
        _a							=	self.vae.decode(a_	).sample

        _a							=	(image / 2 + 0.5).clamp(0     ,     1	)

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

        if output_type == "pil":
          _a							=	self.numpy_to_pil(a_	)

        if not return_dict:
          return image

        return StableDiffusionPipelineOutput(images=a_     ,     nsfw_content_detected=a_	)


